Python Celery任务队列在复杂业务场景中，经常需要处理多个关联任务的顺序执行问题，同时要保证任务不阻塞主线程，这时候就需要结合链式任务和异步执行能力来实现。链式任务可以让多个任务按预设顺序依次执行，前一个任务的返回结果会自动传递给下一个任务作为输入，而异步执行则能让任务在后台运行，不影响主程序的正常流程。
Celery基础环境配置
在使用链式任务和异步执行功能前，需要先完成Celery的基础配置，这里选择Redis作为消息中间件和结果存储后端。
安装依赖
首先通过pip安装需要的依赖包：
pip install celery redis
初始化Celery实例
创建celery_app.py文件，完成Celery实例的初始化配置：
from celery import Celery
# 创建Celery实例，指定应用名称和消息中间件地址
app = Celery(
"task_demo",
broker="redis://127.0.0.1:6379/0",
backend="redis://127.0.0.1:6379/1"
)
# 可选配置，设置任务结果过期时间为1小时
app.conf.result_expires = 3600
异步任务的基本实现
异步执行是Celery的核心特性，定义任务时只需要给函数添加@app.task装饰器，调用任务时使用delay或者apply_async方法即可实现异步执行。
定义简单异步任务
在tasks.py文件中定义两个简单的异步任务：
from celery_app import app
@app.task
def add(a, b):
# 模拟耗时计算
import time
time.sleep(2)
return a + b
@app.task
def multiply(a, b):
import time
time.sleep(2)
return a * b
调用异步任务
在调用脚本中触发异步任务，不会阻塞当前线程：
from tasks import add, multiply
# 异步调用add任务，立即返回任务对象，不会等待任务执行完成
task_result = add.delay(3, 5)
print(f"任务ID: {task_result.id}")
print(f"任务是否完成: {task_result.ready()}")
# 等待任务完成并获取结果
result = task_result.get()
print(f"任务结果: {result}")
链式任务的实现方式
链式任务指的是多个任务按照顺序依次执行，前一个任务的返回结果会作为下一个任务的第一个参数传入，Celery提供了chain方法来实现这个功能。
使用chain构建链式任务
修改调用脚本，实现先执行add任务，再执行multiply任务的链式逻辑：
from celery import chain
from tasks import add, multiply
# 构建链式任务：先执行add(3,5)，得到的结果作为第一个参数传给multiply，第二个参数是4
# 最终执行逻辑为 multiply(add(3,5), 4) = multiply(8,4) = 32
task_chain = chain(add.s(3, 5), multiply.s(4))
chain_result = task_chain.apply_async()
# 获取链式任务的最终结果
final_result = chain_result.get()
print(f"链式任务最终结果: {final_result}")
链式任务的参数传递说明
这里需要说明.s()方法的作用，它是Celery的签名方法，用于固定任务的参数，链式执行时，前一个任务的返回值会自动作为下一个签名任务的第一个参数，所以上面示例中multiply的签名是multiply.s(4)，实际执行时第一个参数是add的返回结果8，第二个参数是固定的4。
链式任务与异步执行结合
链式任务本身也是异步执行的，不会阻塞主线程，我们可以在链式任务执行的同时处理其他逻辑，也可以通过任务ID查询链式任务的执行状态。
from celery import chain
from tasks import add, multiply
import time
# 启动链式任务
task_chain = chain(add.s(2, 3), multiply.s(10))
result = task_chain.apply_async()
print(f"链式任务ID: {result.id}")
print("主线程继续执行其他逻辑...")
time.sleep(1)
# 查询任务状态
print(f"任务是否完成: {result.ready()}")
if result.ready():
print(f"任务结果: {result.get()}")
else:
print("任务仍在执行中")
常见问题与注意事项
- 链式任务中如果某个任务执行失败，整个链会中断，后续任务不会再执行，需要做好任务的异常捕获和重试机制。
- 消息中间件和结果存储后端需要保持可用，否则会导致任务无法正常提交或者结果无法获取。
- 不要在链式任务中传递过大的数据，避免占用过多消息中间件存储空间，影响执行效率。
- 调用
get()方法时会阻塞当前线程直到任务完成，如果不需要立即获取结果，可以不调用这个方法，后续通过任务ID查询结果即可。
总结
通过Celery的异步执行能力，我们可以将耗时的任务放到后台运行，提升主程序的响应速度。而链式任务功能可以优雅地解决多个关联任务的顺序执行问题，避免手动处理任务结果的传递逻辑。实际开发中可以根据业务需求灵活组合这两种特性，应对复杂的任务调度场景。