如何在Django中调度脚本定时执行？附延迟邮件发送示例

针对你提到的「用户提交表单后延迟48小时发送邮件」的场景，下面是几种实用的实现方案：

1. Celery + 消息中间件（推荐中大型项目） #

Celery是Python生态中最成熟的异步任务框架，完美支持延迟任务和定时任务，配合Redis或RabbitMQ作为消息中间件，稳定性高。

实现步骤： #

• 安装依赖：

  pip install celery redis
  

• 在项目根目录创建celery.py，配置Celery：

  import os
  from celery import Celery
  
  os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings')
  app = Celery('your_project_name')
  app.config_from_object('django.conf:settings', namespace='CELERY')
  app.autodiscover_tasks()
  

• 在Django的settings.py中添加Celery配置：

  CELERY_BROKER_URL = 'redis://localhost:6379/0'
  CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
  

• 在你的app下创建tasks.py，定义发送邮件的任务：

  from celery import shared_task
  from django.core.mail import send_mail
  
  @shared_task
  def send_contact_email(contact_data):
      send_mail(
          subject='感谢您的联系',
          message=f'您好{contact_data["name"]}，感谢您的留言，我们会尽快回复您。',
          from_email='your_service@email.com',
          recipient_list=[contact_data['email']],
          fail_silently=False,
      )
  

• 在表单处理视图中调用延迟任务：

  from .tasks import send_contact_email
  
  def contact_submit(request):
      if request.method == 'POST':
          contact_data = {
              'email': request.POST.get('email'),
              'name': request.POST.get('name'),
              'message': request.POST.get('message')
          }
          # 延迟48小时执行（单位：秒）
          send_contact_email.apply_async(args=[contact_data], countdown=48*3600)
          return HttpResponse('提交成功，我们会在48小时内给您发送邮件')
  

• 启动Celery Worker：

  celery -A your_project_name worker --loglevel=info
  

2. Django Q（轻量集成方案，适合中小型项目） #

Django Q是专为Django设计的轻量任务队列，配置简单，无需复杂的中间件管理，支持延迟任务。

实现步骤： #

• 安装依赖：

  pip install django-q
  

• 在settings.py中添加配置：

  INSTALLED_APPS = [
      # ... 其他app
      'django_q',
  ]
  
  Q_CLUSTER = {
      'name': 'DjangoQ',
      'workers': 4,
      'timeout': 90,
      'retry': 120,
      'queue_limit': 50,
      'bulk': 10,
      'redis': {
          'host': 'localhost',
          'port': 6379,
          'db': 0,
      }
  }
  

• 定义任务函数（可放在app的tasks.py中）：

  from django.core.mail import send_mail
  
  def send_contact_email(contact_data):
      send_mail(
          subject='感谢您的联系',
          message=f'您好{contact_data["name"]}，感谢您的留言。',
          from_email='your_service@email.com',
          recipient_list=[contact_data['email']],
          fail_silently=False,
      )
  

• 在视图中触发延迟任务：

  from django_q.tasks import async_task
  from .tasks import send_contact_email
  
  def contact_submit(request):
      if request.method == 'POST':
          contact_data = {
              'email': request.POST.get('email'),
              'name': request.POST.get('name'),
              'message': request.POST.get('message')
          }
          # 延迟48小时执行（单位：秒）
          async_task('yourapp.tasks.send_contact_email', contact_data, schedule=48*3600)
          return HttpResponse('提交成功')
  

• 启动Django Q集群：

  python manage.py qcluster
  

3. 数据库存储 + Cron（极简方案，适合小型项目） #

如果不想引入任何队列服务，可以把任务存在数据库，用Cron定期扫描并执行。你之前提到Cron无法自动执行，其实是需要在服务器上正确配置Cron任务，系统会自动定期触发脚本。

实现步骤： #

• 创建任务模型（在app的models.py）：

  from django.db import models
  from django.utils import timezone
  
  class ScheduledTask(models.Model):
      TASK_TYPE_CHOICES = [('send_email', '发送联系邮件')]
      task_type = models.CharField(max_length=20, choices=TASK_TYPE_CHOICES)
      task_data = models.JSONField()  # 存储联系人数据
      execute_at = models.DateTimeField()
      is_executed = models.BooleanField(default=False)
      created_at = models.DateTimeField(auto_now_add=True)
  
      class Meta:
          ordering = ['execute_at']
  

• 迁移数据库：

  python manage.py makemigrations && python manage.py migrate
  

• 表单处理时创建任务记录：

  from django.utils import timezone
  from .models import ScheduledTask
  
  def contact_submit(request):
      if request.method == 'POST':
          contact_data = {
              'email': request.POST.get('email'),
              'name': request.POST.get('name'),
              'message': request.POST.get('message')
          }
          execute_at = timezone.now() + timezone.timedelta(hours=48)
          ScheduledTask.objects.create(
              task_type='send_email',
              task_data=contact_data,
              execute_at=execute_at
          )
          return HttpResponse('提交成功')
  

• 编写任务执行脚本（项目根目录下run_scheduled_tasks.py）：

  import os
  import django
  from django.utils import timezone
  from your_app.models import ScheduledTask
  from django.core.mail import send_mail
  
  os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
  django.setup()
  
  def run_tasks():
      now = timezone.now()
      # 筛选未执行且到了执行时间的任务
      tasks = ScheduledTask.objects.filter(is_executed=False, execute_at__lte=now)
      for task in tasks:
          if task.task_type == 'send_email':
              data = task.task_data
              send_mail(
                  subject='感谢您的联系',
                  message=f'您好{data["name"]}，感谢您的留言。',
                  from_email='your_service@email.com',
                  recipient_list=[data['email']],
                  fail_silently=False,
              )
          task.is_executed = True
          task.save()
  
  if __name__ == '__main__':
      run_tasks()
  

• 配置Cron定期执行脚本：
  编辑Cron任务（执行crontab -e），添加以下内容（每小时执行一次）：

  0 * * * * /path/to/your/virtualenv/bin/python /path/to/your/project/run_scheduled_tasks.py
  

  保存后Cron会自动按配置时间执行脚本。

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