使用Django默认邮件后端结合Celery发送邮件失败排查

从你的代码和描述来看，核心问题出在Celery任务的定义和调用方式上，导致任务执行时没有正确获取用户参数——虽然Celery日志显示任务成功，但实际上邮件并没有正确发送到目标地址。

一、问题根源 #

1. 任务类的参数传递错误：你在SendConfirmationEmail类的__init__方法中接收用户参数，但Celery在异步执行任务时，会在Worker端重新实例化这个类，初始化时的参数不会被序列化传递到Worker，所以self.user_name、self.user_email等属性都是None。这会导致send_mail调用时收件人或内容缺失，但因为Django对某些异常场景的兼容（比如空收件人未触发报错），Celery日志显示任务成功，但实际没有发送有效邮件。

2. 任务调用方式错误：在signals.py中，你先实例化了SendConfirmationEmail再调用delay()，但delay()是Celery任务的类方法，应该直接通过任务类调用，而不是实例对象。

二、修复方案 #

推荐使用更简洁的函数式任务（Celery官方更推荐这种方式），或者修正类任务的写法：

方案1：改用函数式任务（更简单易维护）

修改tasks.py：

from celery import shared_task
from django.core.mail import send_mail
from django.template import loader
from django.utils.html import strip_tags
from config.settings import default

@shared_task
def send_confirmation_email(username, user_id, user_hash, user_email):
    # 打印日志确认参数（可选，用于排查）
    import logging
    logger = logging.getLogger(__name__)
    logger.info(f"准备发送确认邮件到：{user_email}，用户名：{username}")
    
    confirm_mail = loader.render_to_string(
        'mail/confirmation.html',
        {'user': username, 'id': user_id, 'hash': user_hash, 'domain': default.SITE_URL}
    )
    text_email = strip_tags(confirm_mail)
    send_mail(
        subject='Confirm Your E-mail',
        message=text_email,
        from_email='no-reply@mysite.com',
        recipient_list=[user_email],
        fail_silently=False,
        html_message=confirm_mail
    )

修改signals.py：

from django.db.models.signals import post_save
from django.dispatch import receiver
from apps.siteuser.models import User
from apps.siteuser.tasks import send_confirmation_email

@receiver(post_save, sender=User)
def create_employee_details(sender, instance, created, **kwargs):
    if created:
        # 直接调用任务的delay方法传递参数
        send_confirmation_email.delay(
            username=instance.first_name,
            user_id=instance.id,
            user_hash=instance.hash,
            user_email=instance.email
        )

方案2：修正类任务的写法（如果坚持使用类任务）

修改tasks.py：

from celery import Task
from django.core.mail import send_mail
from django.template import loader
from django.utils.html import strip_tags
from config.celery import app
from config.settings import default
import logging

logger = logging.getLogger(__name__)

class SendConfirmationEmail(Task):
    # 给任务指定唯一名称，避免自动生成的名称冲突
    name = 'apps.siteuser.tasks.send_confirmation_email'

    def run(self, username, user_id, user_hash, user_email, *args, **kwargs):
        logger.info(f"准备发送确认邮件到：{user_email}，用户名：{username}")
        
        confirm_mail = loader.render_to_string(
            'mail/confirmation.html',
            {'user': username, 'id': user_id, 'hash': user_hash, 'domain': default.SITE_URL}
        )
        text_email = strip_tags(confirm_mail)
        send_mail(
            subject='Confirm Your E-mail',
            message=text_email,
            from_email='no-reply@mysite.com',
            recipient_list=[user_email],
            fail_silently=False,
            html_message=confirm_mail
        )

app.register_task(SendConfirmationEmail())

修改signals.py：

from django.db.models.signals import post_save
from django.dispatch import receiver
from apps.siteuser.models import User
from apps.siteuser.tasks import SendConfirmationEmail

@receiver(post_save, sender=User)
def create_employee_details(sender, instance, created, **kwargs):
    if created:
        # 通过任务类直接调用delay方法
        SendConfirmationEmail.delay(
            username=instance.first_name,
            user_id=instance.id,
            user_hash=instance.hash,
            user_email=instance.email
        )

三、额外排查步骤 #

1. 查看Celery日志中的参数：添加日志后，确认Worker端是否接收到正确的用户邮箱和参数，如果参数为空，检查User模型的first_name、hash等字段是否在创建时正确赋值。
2. 确认Mailtrap记录：登录Mailtrap后台，查看"Sent Emails"标签，确认是否有Celery发送的邮件记录（可能被过滤到其他文件夹）。
3. 验证Celery Worker的Django配置加载：确保你的celery.py文件正确加载了Django配置，比如：

import os
from celery import Celery

# 设置Django配置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.default')

app = Celery('config')
# 从Django配置中读取Celery设置
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现所有app中的tasks.py
app.autodiscover_tasks()

内容的提问来源于stack exchange，提问作者SK. Fazlee Rabby