Celery worker运行时如何动态添加并发送定时调度任务

你动态添加定时任务不生效是两个问题叠加导致的：

• 进程隔离问题：你执行celery -A proj2 worker --beat -E启动服务后，Beat调度进程会在自己的内存空间加载当时的Celery配置和定时任务列表。你后续新开Python Shell导入app实例属于完全独立的进程，修改这个独立实例的beat_schedule配置、调用add_periodic_task方法，都只会修改当前Shell进程里的内存数据，根本不会同步到已经在运行的Beat进程，自然不会触发调度。
• 配置笔误：你写的定时任务配置里用了错误的键名tasks，Celery识别定时任务的正确键名是单数形式task，就算配置能同步到Beat，这个笔误也会导致任务加载失败。

根据你要不要重启Beat服务，选对应方案即可：

方案1：接受重启Beat（无需额外依赖，适合测试/简单场景） #

如果你不介意修改任务后重启服务，直接修正配置笔误，把定时任务定义写在Celery app初始化的代码里，或者统一写到celeryconfig.py的beat_schedule配置项中，重启worker和Beat进程就会生效。
修正后的配置示例：

app.conf.beat_schedule = {
   'bye-every-5': {
       'task': 'proj2.tasks.bye', # 注意键名是task不是tasks，值写任务的完整导入路径
       'schedule': 5.0,
   }
}

你之前启动时写的app.add_periodic_task(10.0, hello)之所以能生效，就是因为这行代码在app启动加载阶段执行，Beat进程启动时能直接读到这部分任务定义。

方案2：完全动态添加（无需重启Beat，生产环境推荐） #

如果需要在服务运行时动态增删定时任务、不想每次改配置都重启，需要把Beat的调度存储从默认的进程内存，换成独立的持久化存储（比如Redis、数据库），Beat会按固定间隔从存储里拉取最新的任务列表，自动识别新增/修改的任务。
以常用的RedBeat（基于Redis的动态调度器）为例，操作步骤：

• 安装依赖：

pip install redbeat

• 修改Celery配置：

# proj2/celeryconfig.py
broker_url = 'redis://localhost:6379/0' # 替换成你自己的broker地址
result_backend = 'redis://localhost:6379/0'
# 指定RedBeat作为调度器
beat_scheduler = 'redbeat.RedBeatScheduler'
redbeat_redis_url = 'redis://localhost:6379/0' # 和broker用同一个Redis即可

• 重启一次worker和Beat服务，之后就可以在任意能连通Redis的Python脚本/Shell里动态添加任务，不需要重启服务：

from proj2.celery import app
from celery.schedules import schedule

# 动态添加5秒执行一次的bye任务
app.send_task('redbeat:add_task', kwargs={
    'name': 'bye-every-5',
    'task': 'proj2.tasks.bye',
    'schedule': schedule(5.0),
    'enabled': True
})

添加完成后等1-2个调度周期，Beat就会自动拉取到新任务并开始调度执行。

提示：如果用的是Django技术栈，也可以直接用django-celery-beat扩展，把定时任务存在数据库里，通过Django admin后台就能可视化增删改定时任务，同样不需要重启Beat。

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