在Django项目里实现邮件发送功能,核心是通过内置的邮件模块结合SMTP服务完成,同时为了避免邮件发送阻塞主线程影响接口响应速度,还需要搭配异步任务机制。下面从基础配置到异步实现逐步讲解完整方案。

一、Django SMTP后端基础配置
首先需要在Django的配置文件settings.py中添加邮件相关的配置项,指定SMTP服务器的地址、端口、认证信息等参数。以下是通用配置示例:
# settings.py 邮件配置 EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend' # 使用SMTP后端 EMAIL_HOST = 'smtp.ipipp.com' # SMTP服务器地址,这里替换成实际的服务商地址 EMAIL_PORT = 465 # SMTP端口,SSL通常用465,TLS用587 EMAIL_USE_SSL = True # 是否使用SSL加密 EMAIL_HOST_USER = 'your_email@ipipp.com' # 发件人邮箱 EMAIL_HOST_PASSWORD = 'your_email_password' # 发件人邮箱授权码,不是登录密码 DEFAULT_FROM_EMAIL = EMAIL_HOST_USER # 默认发件人
注意如果是使用TLS加密,需要将EMAIL_USE_SSL改为EMAIL_USE_TLS = True,同时端口调整为587。
二、同步发送邮件的实现
配置完成后可以直接使用Django内置的send_mail函数发送邮件,以下是同步发送的简单示例:
from django.core.mail import send_mail
from django.conf import settings
def send_sync_email():
# 收件人列表
recipient_list = ['test@ipipp.com']
# 发送邮件
send_mail(
subject='Django测试邮件', # 邮件主题
message='这是一封通过Django SMTP后端发送的测试邮件', # 邮件正文
from_email=settings.DEFAULT_FROM_EMAIL, # 发件人
recipient_list=recipient_list, # 收件人列表
fail_silently=False # 发送失败是否抛出异常
)
这种方式是同步执行的,如果邮件发送耗时较长,会阻塞当前请求的处理流程,不适合在高并发场景使用。
三、异步发送邮件的实现
为了避免邮件发送阻塞主线程,我们可以结合Celery异步任务队列实现邮件的异步发送,步骤如下:
3.1 安装依赖
首先需要安装Celery和Redis作为消息中间件:
pip install celery redis
3.2 配置Celery
在项目根目录创建celery.py文件,配置Celery实例:
# 项目根目录/celery.py
import os
from celery import Celery
# 设置Django环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
app = Celery('your_project')
# 从Django配置中读取Celery配置
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现所有app下的任务
app.autodiscover_tasks()
同时在settings.py中添加Celery相关配置:
# settings.py Celery配置 CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0' # 消息中间件地址 CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0' # 结果存储地址 CELERY_TIMEZONE = 'Asia/Shanghai' # 时区
3.3 定义异步邮件任务
在某个app下创建tasks.py文件,定义发送邮件的异步任务:
# app/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from django.conf import settings
@shared_task
def send_async_email(subject, message, recipient_list):
"""
异步发送邮件的任务
:param subject: 邮件主题
:param message: 邮件正文
:param recipient_list: 收件人列表
"""
try:
send_mail(
subject=subject,
message=message,
from_email=settings.DEFAULT_FROM_EMAIL,
recipient_list=recipient_list,
fail_silently=False
)
return '邮件发送成功'
except Exception as e:
# 这里可以添加重试逻辑或者错误日志记录
return f'邮件发送失败: {str(e)}'
3.4 调用异步任务
在视图函数或者其他业务逻辑中调用异步任务即可,不会阻塞当前流程:
# views.py
from django.http import JsonResponse
from .tasks import send_async_email
def user_register(request):
# 假设这里是用户注册逻辑
# 注册成功后异步发送验证邮件
send_async_email.delay(
subject='用户注册验证',
message='请点击链接完成注册验证',
recipient_list=['new_user@ipipp.com']
)
return JsonResponse({'code': 0, 'msg': '注册成功,验证邮件已发送'})
四、常见问题排查
- 如果发送邮件提示认证失败,检查
EMAIL_HOST_PASSWORD是否是邮箱的授权码,不是登录密码,需要在邮箱设置中开启SMTP服务获取授权码。 - 如果连接SMTP服务器超时,检查服务器地址、端口是否正确,以及服务器防火墙是否开放对应端口。
- 异步任务不执行的话,检查Celery worker是否正常启动,启动命令为
celery -A your_project worker -l info。
注意生产环境中建议给邮件发送任务添加重试机制,比如发送失败自动重试3次,避免临时网络问题导致邮件发送失败。