Celery链式任务中task signature设置的eta参数不生效如何解决

问题根因 #

你遇到的问题由三个常见的Celery使用错误导致：

1. 执行选项传参位置错误：eta属于Celery任务的执行配置，不属于任务本身的业务入参，你将其放在kwargs（你拼写错误写成了starkwargs）中传递，只会被当成普通参数传给task_b，无法触发延迟逻辑，还会引发参数类型错误。
2. 绝对时间计算时机错误：你在定义task_b签名时就计算了datetime.now() + timedelta，这个时间是代码执行到签名定义行的绝对时间，不是task_a执行完成后的时间，如果task_a执行耗时超过你设置的延迟，task_b会立刻执行。
3. 任务链配置范围错误：给整个chain的apply_async传eta，只会延迟整个任务链的启动时间，也就是只会推迟task_a的执行，无法作用到链中的后续任务。

正确实现方案 #

方案1：延迟系数可提前确定（不依赖task_a执行结果） #

如果dynamic_coefficient不需要从task_a的返回结果中计算，直接用signature的set()方法配置countdown参数即可，countdown是相对延迟（单位为秒），在任务链中会自动相对于前序任务完成时间生效，完全符合你的需求：

from celery import chain

task_a = task_a.s()
# 延迟分钟转成秒
delay_seconds = dynamic_coefficient * 60
# 给task_b签名设置执行选项
task_b = task_b.s(arg_a).set(countdown=delay_seconds)

# 启动任务链，task_a执行完成后会自动等待对应时间再执行task_b
chain(task_a, task_b).apply_async()

方案2：延迟系数需要动态计算（依赖task_a执行结果） #

如果dynamic_coefficient需要从task_a的返回值中获取，可以加一个无侵入的中间调度任务，不需要修改原有task_a、task_b的代码，也不会造成业务逻辑耦合：

# 新增通用延迟调度任务，不侵入原有业务逻辑
@app.task
def trigger_delayed_task(task_result, dynamic_coefficient, target_task, *task_args, **task_kwargs):
    return target_task.s(*task_args, **task_kwargs).apply_async(countdown=dynamic_coefficient * 60)

# 任务链写法
task_a = task_a.s()
chain(
    task_a,
    trigger_delayed_task.s(dynamic_coefficient, task_b, arg_a)
).apply_async()

报错说明 #

你之前遇到的AttributeError: 'unicode' object has no attribute 'isoformat'错误，正是因为你把eta错误放到了任务业务入参中，Celery内部尝试对执行配置的eta调用isoformat方法时，拿到的是你传入的普通字符串参数，导致类型不匹配报错，用set()方法传递执行选项即可避免该问题。

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