执行长时任务时Celery连接中断，报ConnectionResetError(10054)求助

从你给出的错误日志和Worker启动命令来看，这个问题的核心是Celery Worker在任务执行完成后尝试向AMQP Broker发送ACK确认时，TCP连接已经被远程主机（Broker或中间网络设备）强制断开，下面是具体的排查方向和解决方案：

1. 致命问题：关闭了Celery的心跳机制 #

你的Worker启动命令里包含了--without-heartbeat参数，这直接关闭了Celery与Broker之间的心跳维持机制！

当任务长时间运行时，Worker和Broker之间的TCP连接会处于完全空闲状态，没有任何数据包传输，此时Broker本身、中间的防火墙或负载均衡设备会因为“连接超时”主动断开这条空闲连接。等到任务执行完毕，Worker尝试发送ACK时，就会触发ConnectionResetError。

解决方法： #

• 立即移除--without-heartbeat参数，让Celery默认的心跳机制生效。
• 如果需要自定义心跳间隔，可以在你的Celery配置文件中添加：

  BROKER_HEARTBEAT = 30  # 每30秒发送一次心跳包，可根据你的最长任务时长调整
  

2. 检查AMQP Broker的空闲连接超时配置 #

如果你的Broker是RabbitMQ这类常见的AMQP服务，它本身可能配置了较短的空闲连接超时时间，超过这个时间就会主动断开空闲连接。

解决方法： #

• 查看Broker的配置文件（比如RabbitMQ的rabbitmq.conf），找到连接超时相关的参数：
  • 调整connection_timeout参数，设置为大于你的最长任务运行时间（比如设置为3600，即1小时）。
  • 确保启用了TCP keepalive相关的配置，比如RabbitMQ可以设置：

    tcp_listen_options.nodelay = true
    tcp_listen_options.keepalive = true
    

3. 排查网络层面的空闲连接拦截 #

如果Worker和Broker之间隔着防火墙、负载均衡器或者云厂商的网络组件，这些设备通常会有默认的空闲连接超时规则（比如30分钟），一旦连接空闲超过这个时间就会被强制断开。

解决方法： #

• 联系运维团队调整这些网络设备的空闲超时时间，确保其大于你的最长任务时长。
• 或者在Celery配置中启用TCP keepalive，让操作系统主动维持连接：

  import socket
  
  BROKER_TRANSPORT_OPTIONS = {
      "socket_options": {
          socket.SOL_SOCKET: socket.SO_KEEPALIVE,
          socket.IPPROTO_TCP: socket.TCP_KEEPIDLE, 300,  # 连接空闲5分钟后开始发送keepalive包
          socket.IPPROTO_TCP: socket.TCP_KEEPINTVL, 60, # 每隔1分钟发送一次
          socket.IPPROTO_TCP: socket.TCP_KEEPCNT, 5,     # 连续5次未收到回应则断开连接
      }
  }
  

4. 验证Gevent池的兼容性与配置 #

你使用了-P gevent作为Worker池，需要确保gevent的socket monkey-patching正确生效，否则可能会导致TCP连接的异常处理问题。

解决方法： #

• 在你的Celery应用入口文件（src.celery_app）最顶部添加gevent的monkey patch代码：

  from gevent import monkey
  monkey.patch_all()
  

• 检查gevent、Celery、kombu的版本兼容性，避免因版本不匹配导致的连接问题（建议使用Celery官方推荐的版本组合）。

5. 调整任务ACK的时机（可选权衡方案） #

默认情况下，Celery会在任务执行完成后才向Broker发送ACK确认。如果你的任务运行时间过长，连接很可能在任务完成前就已经断开。

解决方法（需权衡风险）： #

• 在Celery配置中开启acks_late，让Worker在任务开始执行前就发送ACK：

  task_acks_late = True
  

  注意：这个配置的代价是，如果Worker在任务执行过程中崩溃，Broker不会重新分发该任务（因为已经收到ACK），所以只适合可以容忍任务丢失的场景，或者结合任务重试机制使用。

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