8 定时任务与周期性调度

在构建复杂的分布式系统时，我们经常会遇到需要“定时”或“周期性”执行的任务。比如，每天凌晨生成报表，每小时同步一次数据，或者在特定时间发送提醒邮件。这些任务如果都依赖人工触发，不仅效率低下，而且容易出错。

这时，Celery 的定时任务与周期性调度功能就如同你的任务管家，能够精准、可靠地在预设时间执行你的任务。今天，我们就来深入浅出地聊聊 Celery 在这方面的强大能力，包括如何使用 celery beat、如何避免集群中的重复调度，以及如何动态修改定时任务。

一、celery beat：你的任务调度器

celery beat 是 Celery 的一个独立服务，它负责读取你的调度配置，并根据这些配置将任务发送到 Celery 消息队列中，然后由 Celery worker 消费并执行。你可以把它想象成一个拥有智能日历的秘书，到了指定时间就会提醒你的 worker 去完成任务。

1. 基本使用

要使用 celery beat，你需要在 Celery 配置中定义调度规则。Celery 支持多种调度方式：

• crontab 表达式（类似 Linux 的 Cron）：最常见也最强大的方式，可以精细控制任务的执行时间。

  Python

  # celeryconfig.py
  from celery.schedules import crontabCELERY_BEAT_SCHEDULE = {'add-every-monday-morning': {'task': 'tasks.add',  # 你的任务函数路径'schedule': crontab(hour=7, minute=30, day_of_week=1), # 每周一早上7点30分执行'args': (16, 16),},
  }
  

• timedelta 对象（相对时间）：适合设置固定间隔的任务。

  Python

  # celeryconfig.py
  from datetime import timedeltaCELERY_BEAT_SCHEDULE = {'add-every-30-seconds': {'task': 'tasks.add','schedule': timedelta(seconds=30), # 每30秒执行一次'args': (16, 16),},
  }
  

• solar 表达式（日出日落）：适合与地理位置相关的任务。

  Python

  # celeryconfig.py
  from celery.schedules import solarCELERY_BEAT_SCHEDULE = {'send-newsletter-at-sunrise': {'task': 'tasks.send_newsletter','schedule': solar('sunrise', -37.8175306, 144.9657062), # 墨尔本日出时执行'args': ('daily-newsletter',),},
  }
  

配置好调度规则后，只需启动 celery beat 服务：

Bash

celery -A your_project beat

同时，你的 Celery worker 也需要启动：

Bash

celery -A your_project worker -l info

这样，celery beat 就会在指定时间将任务发送到消息队列，worker 就会消费并执行。

二、集群中避免重复调度：让任务有且仅有一次

在生产环境中，为了提高可用性和负载均衡，我们通常会部署多个 Celery worker 实例。那么问题来了：如果部署了多个 celery beat 实例，它们会不会同时调度同一个任务，导致任务被重复执行呢？

答案是：默认情况下会的！这是非常危险的，比如重复发送邮件、重复扣款等。所以，在集群环境下，我们必须采取措施避免重复调度。常见的策略有：

1. Redis 锁（推荐）

使用 Redis 锁是一种轻量且高效的方式。其核心思想是，在任务调度前，celery beat 尝试获取一个针对该任务的分布式锁。只有成功获取锁的 celery beat 实例才能调度任务，其他实例则跳过。

实现思路：

1. 定义一个自定义的 beat 调度器：继承 Celery 提供的 Scheduler 类。
2. 在调度任务前尝试获取 Redis 锁：可以使用 redis-py 库的 setnx 命令。
3. 获取锁成功则调度任务，并设置锁的过期时间：确保锁不会永久占用。
4. 获取锁失败则跳过调度。

伪代码示例：

Python

import redis
from celery.beat import Scheduler
from celery.app.defaults import DEFAULT_BEAT_DB_FILENAMEclass RedisScheduler(Scheduler):def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)self.redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)self.lock_key_prefix = 'celery-beat-lock:'def tick(self, *args, **kwargs):# 遍历所有待调度的任务for name, entry in self.schedule.items():lock_key = f"{self.lock_key_prefix}{name}"# 尝试获取锁，设置过期时间（例如10秒，根据任务执行时间调整）if self.redis_client.set(lock_key, 1, nx=True, ex=10):try:# 成功获取锁，调度任务self.apply_entry(entry, producer=self.producer)self.logger.debug(f"Task '{name}' scheduled by this beat instance.")finally:# 任务调度完成后释放锁（或者等待过期）# self.redis_client.delete(lock_key) # 如果任务执行很快，可以在此释放passelse:self.logger.debug(f"Task '{name}' already locked, skipping.")return super().tick(*args, **kwargs)# 在 Celery 配置中指定使用自定义调度器
# CELERY_BEAT_SCHEDULER = 'your_project.beat_scheduler.RedisScheduler'

2. 数据库记录

另一种方法是使用数据库来记录任务的调度状态。在任务调度前，查询数据库，如果该任务已经被其他 celery beat 实例调度过，则跳过。

实现思路：

1. 在数据库中创建一个表：记录任务名称、上次调度时间、调度实例 ID 等信息。
2. 在调度任务前，查询数据库：检查该任务是否在短时间内被调度过。
3. 如果未被调度，则更新数据库记录并调度任务。
4. 如果已被调度，则跳过。

这种方法相对于 Redis 锁来说，会增加数据库的压力，但对于一些对数据一致性要求更高的场景，或者已经有数据库依赖的系统，也是一个可行的方案。

三、动态修改定时任务：灵活应变

有时候，我们不仅仅需要预设好定时任务，还可能需要在系统运行时动态地添加、修改或删除定时任务，例如：

• 用户自定义定时发送消息。
• 根据业务需求调整报表生成频率。
• 紧急情况下暂停某个周期性任务。

此时，Celery 提供的 django-celery-beat 扩展就显得尤为强大。

django-celery-beat 扩展

django-celery-beat 是一个为 Django 用户量身定制的 Celery Beat 调度器扩展。它将 Celery 的调度信息存储在 Django 的数据库中，并提供了 Django Admin 界面，让你能够非常方便地管理定时任务。

主要特性：

1. 数据库存储调度信息：所有的定时任务配置都存储在 Django 数据库中，方便管理和持久化。
2. Django Admin 集成：通过友好的 Django Admin 界面，你可以直观地添加、编辑、删除定时任务，无需修改代码或重启服务。
3. 支持多种调度类型：包括 crontab、interval（类似 timedelta）等。
4. 动态更新：celery beat 会定期从数据库中加载最新的调度信息，实现任务的动态修改。

使用步骤（简要）：

1. 安装 django-celery-beat：

   Bash

   pip install django-celery-beat
   

2. 添加到 INSTALLED_APPS：

   Python

   # settings.py
   INSTALLED_APPS = [# ...'django_celery_beat',
   ]
   

3. 运行数据库迁移：

   Bash

   python manage.py migrate
   

4. 配置 Celery 使用 django-celery-beat 调度器：

   Python

   # settings.py 或 celery.py
   CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
   

5. 启动 celery beat：

   Bash

   celery -A your_project beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler
   

   或者在 Django 项目根目录下直接运行：

   Bash

   python manage.py celery beat -l info
   

现在，登录你的 Django Admin 界面，你会看到 Periodic Tasks 相关的模型。你可以轻松地在这里创建新的周期性任务，指定任务函数、调度频率、参数等。celery beat 会自动检测到这些变化并应用。

总结

Celery 的定时任务与周期性调度功能是构建健壮分布式系统不可或缺的一部分。通过 celery beat，我们可以灵活地定义各种调度规则。而在集群环境中，通过 Redis 锁或数据库记录等方式，能够有效避免任务的重复调度，确保任务的有且仅有一次执行。最后，借助 django-celery-beat 这样的扩展，我们可以实现对定时任务的动态管理，极大地提高了系统的灵活性和可维护性。

掌握这些知识，你就能更好地驾驭 Celery，让你的后台任务管理更加高效和可靠！