Celery搭配eventlet/gevent无法正常执行任务求助

问题详情 #

我通过代码手动启动Celery Worker，代码如下：

if __name__ == '__main__':
    worker = celery.Worker()
    worker.setup_defaults(
        loglevel=logging.INFO,
        pool='eventlet',
        concurrency=500
    )
    worker.start()

运行后Worker能正常接收任务，但任务一直处于待执行状态；按下CTRL+C触发warm shutdown时，eventlet池下的任务才会批量执行，且执行数量和concurrency参数一致。换成gevent池后问题同样存在，停止脚本后所有任务会被放回队列。但使用prefork池时，任务执行完全正常。

我已经尝试禁用mingle、heartbeat和gossip，问题仍未解决。当前eventlet版本为0.33.2，gevent版本为22.10.2，都是最新版本。想知道这是什么原因？Celery和eventlet/gevent到底兼容吗？

原因分析与解决方法 #

核心问题 #

手动实例化celery.Worker并调用setup_defaults的方式，跳过了Celery针对协程池（eventlet/gevent）的关键初始化流程——没有正确启动协程池的事件循环，也没提前注入猴子补丁，导致任务被接收后无法被调度执行，直到shutdown时才触发批量处理。

具体修复步骤 #

1. 提前注入协程猴子补丁
   在启动Worker前，必须先执行对应协程库的猴子补丁，把所有同步IO操作替换为协程兼容的版本：

   • 用eventlet的话，添加：

     import eventlet
     eventlet.monkey_patch()
     

   • 用gevent的话，添加：

     from gevent import monkey
     monkey.patch_all()
     

2. 改用官方推荐的Worker启动方式
   不要直接手动创建celery.Worker实例，而是通过Celery应用的worker_main方法启动，这种方式会自动完成协程池的全部初始化流程：

   import logging
   import eventlet
   eventlet.monkey_patch()  # 若用gevent则替换为对应补丁代码
   
   from celery import Celery
   
   # 先初始化Celery应用
   app = Celery('your_task_app', broker='your_broker_address')
   
   if __name__ == '__main__':
       app.worker_main(
           argv=[
               'worker',
               '--loglevel=INFO',
               '--pool=eventlet',
               '--concurrency=500'
           ]
       )
   

补充说明 #

• Celery官方完全支持eventlet和gevent池，但必须严格遵循协程池的使用规范：先打补丁，再启动Worker。
• 禁用mingle、heartbeat等配置和这个问题无关，所以你的尝试无法解决核心问题。

内容的提问来源于stack exchange，提问作者Cyril N.