Python不使用Django框架时如何通过Celery发送邮件

不用依赖Django，最小化配置就能把你现有能跑通的email.py接入Celery异步队列，步骤如下：

前置依赖安装 #

• 执行命令安装所需包：pip install celery redis
• 本地启动Redis服务，用默认6379端口即可，不需要额外改配置（如果习惯用RabbitMQ当消息队列也可以，后续只需要改对应broker连接地址，其他代码逻辑不变）

最简项目结构 #

不用搞复杂的目录分层，和你现有email.py平级放3个文件就行：

your_project/
├── celery_app.py  # Celery实例初始化配置
├── email.py       # 你原有已跑通的邮件发送脚本
└── run_task.py    # 业务侧触发邮件任务的入口

编写Celery初始化配置（celery_app.py） #

不需要Django那套settings配置，直接初始化实例即可：

from celery import Celery

# 初始化Celery实例
app = Celery(
    'email_task_queue',
    broker='redis://127.0.0.1:6379/0',  # 任务队列存储地址
    backend='redis://127.0.0.1:6379/1'  # 任务执行结果存储地址
)

# 基础配置
app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Asia/Shanghai',
    enable_utc=True,
    task_time_limit=30,  # 单个邮件任务最长执行30秒，避免阻塞队列
    task_default_retry_delay=10,  # 失败默认重试间隔10秒
)

# 自动扫描当前目录下的任务定义
app.autodiscover_tasks(['.'])

改造原有email.py为Celery任务 #

核心发信逻辑完全不用改，只需要给发信函数加一个Celery任务装饰器即可：

from celery_app import app
# 以下导入你原有发信需要的依赖，比如smtplib、邮件模板类等，和之前保持一致
import smtplib
from email.mime.text import MIMEText

# 加装饰器把普通函数转为Celery异步任务，配置最多重试3次
@app.task(bind=True, max_retries=3)
def send_email(self, to_addr: str, subject: str, content: str):
    try:
        # 这里完全保留你之前已经验证可正常运行的发信代码，不需要修改
        msg = MIMEText(content, 'plain', 'utf-8')
        msg['From'] = '你的发件邮箱地址'
        msg['To'] = to_addr
        msg['Subject'] = subject

        smtp = smtplib.SMTP_SSL('你的SMTP服务地址', 465)
        smtp.login('你的发件邮箱地址', '邮箱授权码')
        smtp.sendmail('你的发件邮箱地址', to_addr, msg.as_string())
        smtp.quit()
        return f"邮件发送成功，收件人：{to_addr}"
    except Exception as e:
        # 发信失败自动重试
        self.retry(exc=e)

启动Celery工作进程 #

在项目根目录打开终端，执行启动命令：
celery -A celery_app worker --loglevel=info

Windows环境如果启动报错，追加--pool=solo参数即可正常运行。每次修改任务代码后必须重启worker，否则新代码不会生效。

业务侧调用异步任务 #

在你需要触发发信的业务代码里，替换原来的同步调用为异步投递即可：

from email import send_email

# 原来的同步调用：程序会阻塞等待邮件发完才执行后续逻辑
# send_email("test@example.com", "测试邮件", "这是邮件内容")

# 异步调用：任务直接投递到队列，worker后台执行，主程序立刻往下走
task_res = send_email.delay("test@example.com", "测试邮件", "这是邮件内容")
# 需要的话可以打印任务ID，后续用来查询发送状态
print(f"邮件任务已提交，任务ID：{task_res.id}")

常见问题排查 #

• 出现循环导入报错：检查是不是把Celery实例初始化和任务定义写在了同一个文件里，按上面的结构单独拆分celery_app.py即可解决。
• 任务一直处于待执行状态：检查Redis服务是否正常启动、worker进程是否正常运行、任务参数是否能被JSON序列化（不要传不能转JSON的对象比如文件句柄、数据库连接当任务参数）。
• 需要发送HTML邮件、带附件的邮件：直接修改send_email函数里的邮件拼接逻辑即可，和你之前写同步代码的逻辑完全一致，Celery只负责任务调度，不会侵入业务逻辑。

内容的提问来源于stack exchange，提问作者user11663746