如何在Celery延迟任务中执行apply_async任务？遇属性错误求助

这个错误的核心原因很清晰：你传给TaskA的func是个普通Python函数（也就是TaskB），但apply_async是Celery task对象独有的方法，普通函数当然没有这个属性。

要解决这个问题，只需要两步：

• 把TaskB标记为Celery task：用@task装饰器修饰它，让Celery把它识别为可异步调度的task对象。
• 确保传递的是task对象：装饰后的TaskB本身就是task对象，直接传递即可。

修改后的完整代码如下：

from celery import task

def Test(func, delay_seconds):
    TaskA.apply_async((func, delay_seconds), countdown=1)

@task
def TaskA(func, delay_seconds):
    result = xxx  # 这里保留你原本的判断逻辑
    if result:
        func.apply_async(countdown=delay_seconds)

@task  # 关键：给TaskB加上Celery task装饰器
def TaskB():
    xxx  # 你的TaskB业务逻辑

# 调用时直接传递TaskB这个task对象
Test.apply_async((TaskB, 600,), countdown=60)

额外补充个小提示：如果你不想把TaskB变成Celery task，也可以在TaskA里直接调用func()，但这样是同步执行的——会在TaskA的worker进程里直接运行TaskB，而不是异步调度新的任务，这可能不符合你原本的需求。

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