Python分布式任务队列Celery收到任务后不执行，如何排查和解决？

遇到Celery任务不执行，先别慌，按这个顺序查，基本能搞定。

第一步：确认基础配置和连接 先看你的Celery应用配置和Redis/RabbitMQ连接。跑下面这个脚本，能帮你快速验证：

# test_celery_basic.py
from your_celery_app import app  # 替换为你的实际导入

# 检查配置
print("Broker URL:", app.conf.broker_url)
print("Result Backend:", app.conf.result_backend)

# 测试连接
with app.connection() as conn:
    conn.ensure_connection(max_retries=3)
    print("Broker连接成功！")

# 发送一个最简单的测试任务
@app.task
def test_task():
    return "任务执行成功！"

# 异步调用并获取结果
result = test_task.delay()
print("任务ID:", result.id)
print("任务结果（阻塞获取）:", result.get(timeout=10))

如果这里就报连接错误，检查broker_url（比如Redis是redis://localhost:6379/0）和消息队列服务是否真的在运行。

第二步：检查Worker状态 任务发了但没执行，八成是Worker没干活。启动Worker时一定要指定正确的应用模块和队列：

# 在项目根目录下执行，确保能正确导入你的Celery app实例
celery -A your_celery_app_module:app worker --loglevel=info -Q your_queue_name

在日志里看到Connected to和Ready to accept tasks才算成功。如果没日志，加--loglevel=debug再启动。

第三步：验证任务路由和序列化 有时候任务发错队列或者参数无法序列化。在发送任务的代码里加个检查：

# 在调用任务的地方附近添加
task_result = your_task.delay(*args, **kwargs)
print(f"任务已发送，ID: {task_result.id}, 状态: {task_result.state}")

# 或者直接检查
from celery.exceptions import OperationalError
try:
    your_task.apply_async(args=[...], queue='your_queue')
except OperationalError as e:
    print(f"发送失败: {e}")

第四步：同步调用测试 在排查时，可以先用apply()同步执行，排除异步调度的问题：

# 临时测试：同步执行
result = your_task.apply(args=[...], kwargs={...})
print("同步执行结果:", result.get())

如果同步能成但异步不行，问题肯定在Broker或Worker。

最后快速过一遍常见坑点：

1. 模块导入问题：启动Worker的目录不对，导致找不到Celery app实例。
2. Windows系统：4.0+版本对Windows支持不好，用eventlet或换Linux测试。
3. 任务注册：确保装饰器@app.task用在正确的地方，或者用app.register_task()手动注册。
4. 防火墙/网络：检查Broker端口（Redis的6379，RabbitMQ的5672）是否可达。

总结建议：从Worker日志和连接状态开始查起。