如何使用Celery根据模型due_date字段向用户发送到期提醒邮件

核心思路是双层任务兜底+幂等校验，完全避免漏发、重复发、时间不匹配的问题，不需要给每个任务单独建定时规则，维护成本低。

第一步：补全模型必要字段 #

你现有的Task模型缺几个必要字段，先加上，否则没法做防重、没法关联收件人：

from django.db import models
from django.contrib.auth import get_user_model

User = get_user_model()

class Task(BaseModel):
    name = models.CharField(max_length=255)
    due_date = models.DateField(blank=True, null=True)
    # 以下是新增字段
    owner = models.ForeignKey(User, on_delete=models.CASCADE, related_name="tasks", verbose_name="任务所属人")
    reminder_sent = models.BooleanField(default=False, verbose_name="到期提醒已发送")
    celery_reminder_task_id = models.CharField(max_length=255, blank=True, null=True, verbose_name="预调度的Celery任务ID")

加完记得执行python manage.py makemigrations && python manage.py migrate同步表结构。

第二步：基础配置对齐 #

先把Celery和Django的时区配置统一，否则会出现日期计算差一天的问题：

• Django配置里的TIME_ZONE设为你业务用的时区，比如Asia/Shanghai
• Celery配置里加CELERY_TIMEZONE = "Asia/Shanghai"，CELERY_ENABLE_UTC = False
• 所有时间计算统一用Django自带的django.utils.timezone工具，不要用原生datetime.now()。

第三步：实现两个核心Celery任务 #

两个任务配合，一个负责实际发邮件，一个负责每日巡检兜底，哪怕预调度的任务丢了、due_date被改了，也绝对不会漏发。

1. 原子发送任务（幂等） #

这个任务只做一件事：校验任务状态，符合条件就发邮件，不管被谁调用，都不会重复发、不会错发：

from celery import shared_task
from django.core.mail import send_mail
from django.conf import settings
from django.utils import timezone
from .models import Task
import logging

logger = logging.getLogger(__name__)

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def send_task_due_reminder(self, task_id):
    # 先查未发提醒的任务，不存在直接跳过
    try:
        task = Task.objects.select_related("owner").get(id=task_id, reminder_sent=False)
    except Task.DoesNotExist:
        logger.info(f"任务{task_id}不存在或已发提醒，跳过")
        return

    # 二次校验到期日，避免调度时间偏差
    current_date = timezone.localdate()
    if task.due_date != current_date:
        logger.info(f"任务{task_id}到期日为{task.due_date}，当前为{current_date}，暂不发送")
        return

    # 实际发邮件
    try:
        send_mail(
            subject=f"提醒：您的任务「{task.name}」今日到期",
            message=f"任务名称：{task.name}\n到期日期：{task.due_date}\n请及时处理。",
            from_email=settings.DEFAULT_FROM_EMAIL,
            recipient_list=[task.owner.email],
            fail_silently=False,
        )
        # 发完标记已发送
        task.reminder_sent = True
        task.save(update_fields=["reminder_sent"])
        logger.info(f"任务{task_id}到期提醒已发送至{task.owner.email}")
    except Exception as e:
        logger.error(f"发送任务{task_id}提醒失败，准备重试：{str(e)}")
        self.retry(exc=e)

2. 每日巡检兜底任务 #

这个任务固定每天凌晨0点10分执行，把所有当天到期、还没发提醒的任务全部捞出来，异步触发发送，是防漏发的核心屏障：

@shared_task
def daily_due_task_scan():
    current_date = timezone.localdate()
    # 筛选符合条件的任务：今日到期、未发提醒、有所属人、到期日非空
    due_tasks = Task.objects.filter(
        due_date=current_date,
        reminder_sent=False,
        owner__isnull=False,
        due_date__isnull=False
    )
    # 逐个触发发送，不阻塞巡检流程
    for task in due_tasks:
        send_task_due_reminder.delay(task.id)
    logger.info(f"每日到期任务巡检完成，共触发{due_tasks.count()}条提醒")

这个定时任务直接在django-celery-beat里配置就行，cron规则设为0 10 0 * * *（每天0点10分），存在数据库里不会因为服务重启丢失。

可选增强：任务保存时动态预调度 #

如果不想等凌晨巡检，想任务一到到期日就立刻发，可以加个模型保存信号，在任务新建/修改时提前把发送任务排到Celery队列里，注意这个只是优化体验，不能替代每日巡检：

from django.db.models.signals import post_save
from django.dispatch import receiver
from celery.result import AsyncResult

@receiver(post_save, sender=Task)
def schedule_reminder_when_task_save(sender, instance, **kwargs):
    # 无到期日/已发提醒，直接跳过
    if not instance.due_date or instance.reminder_sent:
        return
    
    current_date = timezone.localdate()
    # 先撤销之前绑定的旧调度任务，避免due_date修改后旧任务误触发
    if instance.celery_reminder_task_id:
        try:
            AsyncResult(instance.celery_reminder_task_id).revoke()
        except:
            pass
    
    # 到期日就是今天，直接触发发送
    if instance.due_date == current_date:
        celery_task = send_task_due_reminder.delay(instance.id)
        instance.celery_reminder_task_id = celery_task.id
        instance.save(update_fields=["celery_reminder_task_id"])
        return
    
    # 到期日在未来，设置eta为到期日当天早上9点，预排任务
    if instance.due_date > current_date:
        send_time = timezone.datetime.combine(
            instance.due_date,
            timezone.time(9, 0),
            tzinfo=timezone.get_current_timezone()
        )
        celery_task = send_task_due_reminder.apply_async(
            args=[instance.id],
            eta=send_time
        )
        instance.celery_reminder_task_id = celery_task.id
        instance.save(update_fields=["celery_reminder_task_id"])

避坑说明 #

• 不要给每个Task单独建PeriodicTask定时规则，数据量上来之后调度器会被拖垮，维护成本极高。
• 发送任务必须做幂等校验，不管被调度多少次，只要发过就直接跳过，避免用户收到重复邮件。
• 如果你的due_date后来改成精确到时分秒的DateTimeField，逻辑只需要调整两点：巡检频率改成每10分钟跑一次，eta时间直接设为due_date的值即可。

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