3 Celery 定时任务

3.1 基本定时任务

3.1.1 创建定时任务文件

设定时间让 Celery 执行一个定时任务，produce_task.py：

from celery_task import send_email
from datetime import datetime, UTC

# 方式一

# 使用本地时间
v1 = datetime(2025, 11, 19, 22, 17, 0)
print(v1)
# 转成 UTC 时间
v2 = datetime.fromtimestamp(v1.timestamp(), UTC)
print(v2)
result = send_email.apply_async(  # type: ignore
    args=[
        "pangzl",
    ],
    eta=v2,
)
print(result.id)

# 方式二
from docs.project.celery.celery_task import send_email

# 当前时间往后推10s
result = send_email.apply_async(args=["pangzl"], countdown=10)
print(result.id)

3.2 多任务结构中的定时任务

3.2.1 修改主配置文件 celery.py

多任务结构中 celery.py 修改如下：

from datetime import timedelta
from celery import Celery
from celery.schedules import crontab

cel = Celery(
    "tasks",
    broker="redis://127.0.0.1:6379/1",
    backend="redis://127.0.0.1:6379/2",
    include=[
        "celery_tasks.task01",
        "celery_tasks.task02",
    ],
)
cel.conf.timezone = "Asia/Shanghai"
cel.conf.enable_utc = False

# 配置定时任务
cel.conf.beat_schedule = {
    # 名字随意命名
    "add-every-10-seconds": {
        # 执行tasks1下的test_celery函数
        "task": "celery_tasks.task01.send_email",
        # 每隔2秒执行一次
        # 'schedule': 1.0,
        # 'schedule': crontab(minute="*/1"),
        "schedule": timedelta(seconds=6),
        # 传递参数
        "args": ("张三",),
    },
    # 'add-every-12-seconds': {
    #     'task': 'celery_tasks.task01.send_email',
    #     每年4月11号，8点42分执行
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'args': ('张三',)
    # },
}

3.2.2 启动 Beat 和 Worker

启动 Beat 程序：

celery beat -A proj

Celery Beat 进程会读取配置文件的内容，周期性的将配置中到期需要执行的任务发送给任务队列。

之后启动 worker 进程