Django定时通知实现需求及cron任务方案的缺陷

1. Celery动态延迟任务（推荐） #

抛弃固定间隔的cron，改用Celery的动态延迟任务，直接在创建通知时指定执行时间，完全避免轮询和延迟：

首先定义Celery任务（需提前配置Celery和消息中间件如Redis/RabbitMQ）：

# tasks.py
from celery import shared_task
from django.core.mail import send_mail
from .models import ScheduledNotification

@shared_task
def send_scheduled_notification(notification_id):
    notification = ScheduledNotification.objects.get(id=notification_id)
    # 替换成你的实际通知逻辑（邮件/站内信/推送等）
    send_mail(
        subject=notification.title,
        message=notification.content,
        from_email='notify@yourdomain.com',
        recipient_list=[notification.user.email],
        fail_silently=False,
    )
    notification.status = 'sent'
    notification.save()

创建通知时直接调度任务：

# 在视图/信号/业务逻辑中
from datetime import datetime
from .tasks import send_scheduled_notification
from .models import ScheduledNotification

def create_notification(user, title, content, send_at):
    notification = ScheduledNotification.objects.create(
        user=user,
        title=title,
        content=content,
        send_at=send_at,
        status='pending'
    )
    # 计算延迟秒数，到点自动执行
    delay_seconds = (send_at - datetime.now()).total_seconds()
    send_scheduled_notification.apply_async(args=[notification.id], countdown=delay_seconds)

2. APScheduler集成Django #

APScheduler支持日期触发的一次性任务，适合不需要复杂队列的场景：

先安装依赖：pip install apscheduler django-apscheduler

配置调度器：

# scheduler.py
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore
from django.utils import timezone
from .models import ScheduledNotification
from django.core.mail import send_mail

def execute_notification(notification_id):
    notification = ScheduledNotification.objects.get(id=notification_id)
    send_mail(
        subject=notification.title,
        message=notification.content,
        from_email='notify@yourdomain.com',
        recipient_list=[notification.user.email],
        fail_silently=False,
    )
    notification.status = 'sent'
    notification.save()

def start_scheduler():
    scheduler = BackgroundScheduler(timezone=timezone.get_current_timezone())
    scheduler.add_jobstore(DjangoJobStore(), 'default')
    # 启动时加载未发送的任务
    pending = ScheduledNotification.objects.filter(status='pending', send_at__gte=timezone.now())
    for notice in pending:
        scheduler.add_job(
            execute_notification,
            'date',
            run_date=notice.send_at,
            args=[notice.id],
            id=f"notice_{notice.id}"
        )
    scheduler.start()

在apps.py中注册启动：

from django.apps import AppConfig

class NotificationsConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'notifications'

    def ready(self):
        from .scheduler import start_scheduler
        start_scheduler()

新增通知时，直接添加到调度器即可，无需轮询。

3. 优化cron轮询（备选） #

如果必须用cron，就缩小轮询间隔到1分钟，同时只查询当前时间窗口内的任务，减少空跑消耗：

编写Django管理命令：

# management/commands/send_pending_notices.py
from django.core.management.base import BaseCommand
from django.utils import timezone
from datetime import timedelta
from .models import ScheduledNotification
from django.core.mail import send_mail

class Command(BaseCommand):
    help = 'Send notifications scheduled for the current time window'

    def handle(self, *args, **options):
        now = timezone.now()
        # 只处理当前时间前后1分钟的待发送任务
        time_window = timedelta(minutes=1)
        pending = ScheduledNotification.objects.filter(
            status='pending',
            send_at__gte=now - time_window,
            send_at__lte=now + time_window
        )
        for notice in pending:
            send_mail(
                subject=notice.title,
                message=notice.content,
                from_email='notify@yourdomain.com',
                recipient_list=[notice.user.email],
                fail_silently=False,
            )
            notice.status = 'sent'
            notice.save()

设置cron每分钟执行：

* * * * * /path/to/venv/bin/python /path/to/project/manage.py send_pending_notices

这样即使无任务，脚本也会快速结束，延迟最多1分钟。

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