Python Celery 定时任务配置与执行详解
发布时间：2026-10-15| 3人浏览过
本文详细讲解如何在 Python 中使用 Celery 实现定时任务。通过配置 Redis 中间件和 Celery Beat 进程，实现周期性或固定间隔的任务触发。涵盖环境搭建、配置文件编写（如 crontab 和 timedelta 设置）、Worker 与 Beat 启动命令及结果验证，帮助开发者快速掌握 Celery 定时调度机制。
前言
在 Python 开发中，Celery 不仅用于处理异步任务（async task），还支持执行定时任务（beat）。
一、Celery 定时任务概述
Celery 的默认任务单元通常由任务生产者主动触发，但在某些场景下，我们需要任务自动触发。Beat 进程正是负责此类任务的组件，它能够自动触发定时或周期性的任务执行。
实现 Celery 的周期任务相对简单，只需在配置文件中定义好周期任务规则，并运行一个周期任务触发器（即 Beat 服务）即可。具体配置方法可参考相关文档。
二、使用步骤
本教程基于以下环境：Windows 10、Python 3、Redis 以及 Celery 5。
1. 代码结构
主要配置文件为 celery_config.py，其内容如下：
配置代码示例如下：
#-*-coding=utf-8-*-
from datetime import timedelta
from celery.schedules import crontab
# 中间件
BROKER_URL = 'redis://scck2svr03:6379/6'
# 结果存储
# CELERY_RESULT_BACKEND = 'redis://:scck2svr03:6379/5'
CELERYD_CONCURRENCY = 20 # 并发worker数
CELERYD_FORCE_EXECV = True # 非常重要,有些情况下可以防止死锁
# CELERY_TASK_SERIALIZER = 'msgpack' # 任务序列化和反序列化使用msgpack方案
CELERY_RESULT_SERIALIZER = 'json' # 读取任务结果一般性能要求不高，所以使用了可读性更好的JSON
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # celery任务结果有效期
CELERY_ACCEPT_CONTENT = ['json', 'msgpack'] # 指定接受的内容类型
CELERY_TIMEZONE = 'Asia/Shanghai' # celery使用的时区
CELERY_ENABLE_UTC = True # 启动时区设置
# 默认worker队列
#CELERY_DEFAULT_QUEUE = 'default'
# 异步任务
CELERY_IMPORTS = (
"celery_learning.email.tasks"
)
# celery beat
CELERYBEAT_SCHEDULE = {
'periodic_task-every-minute': {
'task': 'celery_learning.email.tasks.add',
'schedule': crontab(minute="*/1"),
'args': (1,2)
},
'periodic_task-every-10second':{
'task':'celery_learning.email.tasks.add',
'schedule':timedelta(seconds=10),
'args':(1,12)
}
}2. 启动定时任务
启动命令示例如下：
终端1:celery -A celery_learning.celery_app worker -l info -P gevent
终端2:celery -A celery_learning.celery_app beat终端 1 用于启动 Worker。
终端 2 用于启动 Beat。
3. 执行结果
配置了两个定时任务：一个每分钟执行一次，另一个每 10 秒执行一次。
运行结果如下所示：
总结
至此，关于 Python Celery 定时任务的详细讲解已完成。如需了解更多 Python Celery 相关内容，请搜索本站历史文章或继续浏览后续文章，感谢大家的支持！
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
