Django异步邮件发送怎么实现

本文目录导读：

1. 使用 django.core.mail + 队列系统
2. HTML邮件和附件
3. 批量异步发送
4. 错误处理和重试机制
5. 邮件服务配置示例
6. 推荐方案

在Django中实现异步邮件发送有多种方法,我来介绍几种常用的方案：

使用 django.core.mail + 队列系统

使用Celery（推荐）

# tasks.py
from celery import shared_task
from django.core.mail import send_mail
@shared_task
def send_email_task(subject, message, from_email, recipient_list):
    send_mail(
        subject,
        message,
        from_email,
        recipient_list,
        fail_silently=False,
    )

配置：

# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.gmail.com'
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_HOST_USER = 'your-email@gmail.com'
EMAIL_HOST_PASSWORD = 'your-password'

使用：

# views.py
from .tasks import send_email_task
def send_email_view(request):
    send_email_task.delay(
        'Hello',
        'This is a test email.',
        'from@example.com',
        ['to@example.com']
    )
    return HttpResponse('Email will be sent!')

使用 django-background-tasks

# tasks.py
from background_task import background
from django.core.mail import send_mail
@background(schedule=10)  # 10秒后执行
def send_email_background(subject, message, from_email, recipient_list):
    send_mail(subject, message, from_email, recipient_list)

安装配置：

pip install django-background-tasks

# settings.py
INSTALLED_APPS = [
    'background_task',
    # ...
]

使用Django的缓存系统作为简单队列

from django.core.cache import cache
from django.core.mail import send_mail
import threading
import time
class AsyncEmailService:
    def __init__(self):
        self.queue_key = 'email_queue'
    def add_to_queue(self, subject, message, from_email, recipient_list):
        email_data = {
            'subject': subject,
            'message': message,
            'from_email': from_email,
            'recipient_list': recipient_list,
        }
        queue = cache.get(self.queue_key, [])
        queue.append(email_data)
        cache.set(self.queue_key, queue, 3600)
        # 启动后台发送线程
        self._process_queue()
    def _process_queue(self):
        def process():
            while True:
                queue = cache.get(self.queue_key, [])
                if queue:
                    email = queue.pop(0)
                    cache.set(self.queue_key, queue, 3600)
                    try:
                        send_mail(
                            email['subject'],
                            email['message'],
                            email['from_email'],
                            email['recipient_list'],
                        )
                    except Exception as e:
                        print(f"Failed to send email: {e}")
                else:
                    break
                time.sleep(1)
        thread = threading.Thread(target=process)
        thread.start()
# 使用
async_email = AsyncEmailService()
async_email.add_to_queue('Subject', 'Message', 'from@example.com', ['to@example.com'])

HTML邮件和附件

# tasks.py
from celery import shared_task
from django.core.mail import EmailMultiAlternatives
from django.template.loader import render_to_string
@shared_task
def send_html_email_task(subject, template_name, context, from_email, recipient_list, attachments=None):
    html_content = render_to_string(template_name, context)
    email = EmailMultiAlternatives(
        subject,
        '',  # plain text content
        from_email,
        recipient_list,
    )
    email.attach_alternative(html_content, "text/html")
    if attachments:
        for attachment in attachments:
            email.attach(*attachment)
    email.send(fail_silently=False)

批量异步发送

# tasks.py
from celery import shared_task, group
from django.core.mail import send_mail
@shared_task
def send_single_email(user_email):
    send_mail(
        'Welcome!',
        'Thank you for registering.',
        'from@example.com',
        [user_email],
        fail_silently=True,
    )
@shared_task
def send_bulk_emails(user_emails):
    job = group([send_single_email.s(email) for email in user_emails])
    job.apply_async()

错误处理和重试机制

# tasks.py
from celery import shared_task
from celery.exceptions import MaxRetriesExceededError
from django.core.mail import send_mail
@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def send_email_with_retry(self, subject, message, from_email, recipient_list):
    try:
        send_mail(subject, message, from_email, recipient_list)
    except Exception as exc:
        raise self.retry(exc=exc)

邮件服务配置示例

# settings.py
# 使用Gmail
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.gmail.com'
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_HOST_USER = 'your-email@gmail.com'
EMAIL_HOST_PASSWORD = 'your-app-password'  # Gmail需要应用专用密码
# 使用SendGrid
SENDGRID_API_KEY = 'your-api-key'
EMAIL_BACKEND = "sendgrid.django.SendGridBackend"
# 使用Amazon SES
EMAIL_BACKEND = 'django_ses.SESBackend'
AWS_ACCESS_KEY_ID = 'your-access-key'
AWS_SECRET_ACCESS_KEY = 'your-secret-key'
AWS_SES_REGION_NAME = 'us-east-1'

推荐方案

对于生产环境，我推荐使用 Celery + Redis/RabbitMQ 方案,因为：

1. 完善的错误处理机制
2. 支持任务调度和重试
3. 可以监控任务状态
4. 支持任务优先级
5. 社区活跃，文档完善

如果项目规模较小，可以考虑使用 django-background-tasks 作为轻量级替代方案。