如何通过Celery的task_annotations为任务设置时间限制

我尝试在Celery应用配置中使用task_annotations来强制任务执行时间限制，根据官方文档示例及说明这应该可行，但实际应用后未生效。测试示例如下：

测试代码（simple.py） #

from celery import Celery, shared_task
from time import sleep

class Config:
#    task_time_limit = 1 # 取消注释后可生效
    task_annotations = {'*': {'task_time_limit': 1}} # 预期此处生效
    
app = Celery('tasks')

app.config_from_object(Config)
    
@shared_task(name='run_task') # 已知可在装饰器中设置task_time_limit
def run_task():
    sleep(5)
    print('Completed without timing out :(')

运行命令 #

• 启动Worker：celery -A simple worker -l info
• 触发任务：python -c "from simple import run_task; run_task.delay()"（需本地运行RabbitMQ）

问题：如何修改配置中的task_annotations使时间限制生效？以及除*外，用什么键可指定给我的run_task任务？

1. 让task_annotations生效的修改 #

方法一：改用@app.task绑定到应用实例 #

将@shared_task替换为@app.task，让任务直接归属当前Celery实例，确保配置被正确加载：

from celery import Celery
from time import sleep

class Config:
    task_annotations = {'*': {'task_time_limit': 1}}
    
app = Celery('tasks')
app.config_from_object(Config)
    
@app.task(name='run_task')
def run_task():
    sleep(5)
    print('Completed without timing out :(')

方法二：为shared_task设置默认应用 #

如果坚持使用shared_task，在配置完成后添加app.set_default()，让shared_task使用当前配置好的应用实例：

from celery import Celery, shared_task
from time import sleep

class Config:
    task_annotations = {'*': {'task_time_limit': 1}}
    
app = Celery('tasks')
app.config_from_object(Config)
app.set_default()  # 将当前应用设为默认，shared_task会使用此实例的配置
    
@shared_task(name='run_task')
def run_task():
    sleep(5)
    print('Completed without timing out :(')

关键注意事项 #

修改配置后必须完全重启Worker，避免旧配置被缓存，重新执行启动命令：celery -A simple worker -l info

2. 指定单个任务的task_annotations键值 #

针对run_task任务，有两种键值可以使用：

• 装饰器中自定义的任务名称：'run_task'
• 任务的完全限定名（模块名+函数名）：'simple.run_task'

示例配置：

class Config:
    task_annotations = {
        # 二选一即可
        'run_task': {'task_time_limit': 1},
        # 'simple.run_task': {'task_time_limit': 1}
    }

内容的提问来源于stack exchange，提问作者Francisco Gama T. R.