Celery+RabbitMQ：任务执行时Worker被移出消费者列表问题求助

我通过以下命令启动以RabbitMQ作为消息代理的Celery Worker：

celery -A my_app worker -l info -P gevent -c 100 --prefetch-multiplier=1 -Q my_app

任务代码如下：

@shared_task(queue='my_app', default_retry_delay=10, max_retries=1, time_limit=8 * 60)
def example_task():
    # getting queryset with some filtering
    my_models = MyModel.objects.filter(...)

    for my_model in my_models.iterator():
        my_model.execute_something()

该任务有时1分钟内完成，高负载时需5分钟以上。核心问题是RabbitMQ会随机将Worker从消费者列表中移除，不得不重启Worker，同时Worker抛出以下错误：

SSLEOFError: EOF occurred in violation of protocol (_ssl.c:2396)

有时还会出现：

consumer: Cannot connect to amqps://my_app:**@example.com:5671/prod: SSLEOFError(8, 'EOF occurred in violation of protocol (_ssl.c:997)').

Couldn't ack 2057, reason:"RecoverableConnectionError(None, 'connection already closed', None, '')"

尝试添加--without-heartbeat参数无效果，部分任务耗时超过30分钟，无法持续监控Worker状态，请问如何解决？

1. 修正心跳配置，禁用--without-heartbeat #

完全禁用心跳会让RabbitMQ判定连接已死直接断开，正确做法是调整心跳参数保持连接活性：

• 在Celery配置文件中添加：

  app.conf.broker_heartbeat = 30
  app.conf.broker_heartbeat_checkrate = 2
  

  30秒心跳间隔、每2秒检查一次，既避免RabbitMQ因无活动断连，又不会产生过多流量。
• 启动Worker时移除--without-heartbeat，改用--heartbeat-interval=30保持参数一致。

2. 适配gevent的SSL环境 #

gevent对SSL的兼容性需要额外配置，解决IO阻塞导致的连接异常：

• 安装依赖：

  pip install gevent-websocket pyopenssl
  

• 启动前设置环境变量：

  export GEVENT_RESOLVER=ares
  export SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt
  

  强制gevent使用pyopenssl处理SSL，避免证书解析或DNS问题引发的EOF错误。

3. 调整Worker并发与连接池配置 #

当前-c 100的并发数结合gevent可能导致连接池过载：

• 降低并发数至-c 50，观察稳定性后再逐步调整。
• 在Celery配置中匹配连接池上限：

  app.conf.broker_pool_limit = 50  # 与-c参数值一致
  

  避免连接池耗尽引发的连接失败。

4. 开启自动重连机制 #

让Worker在连接断开后自动恢复，无需手动重启：

app.conf.broker_connection_retry = True
app.conf.broker_connection_retry_on_startup = True
app.conf.broker_max_retries = 10  # 最大重试次数

5. 优化长任务的连接保持 #

针对耗时超30分钟的任务，避免连接因闲置被断开：

• 拆分长任务为多个短任务，减少单任务占用连接的时间。
• 若无法拆分，在任务中定期刷新连接：

  from celery import current_app
  
  @shared_task(queue='my_app', default_retry_delay=10, max_retries=1, time_limit=8 * 60)
  def example_task():
      my_models = MyModel.objects.filter(...)
      for idx, my_model in enumerate(my_models.iterator()):
          my_model.execute_something()
          # 每处理100条数据刷新一次连接
          if idx % 100 == 0:
              current_app.connection().ensure_connection()
  

6. 检查RabbitMQ端SSL配置 #

确认RabbitMQ的超时设置适配长任务：

• 调整ssl_handshake_timeout和connection_timeout参数，确保值大于任务最长执行时间。
• 验证Worker端信任RabbitMQ证书：

  import ssl
  app.conf.broker_use_ssl = {
      'cert_reqs': ssl.CERT_REQUIRED,
      'ca_certs': '/path/to/your/ca.crt',
  }
  

内容的提问来源于stack exchange，提问作者Mr.D