Python Asyncio生产者消费者模型运行后消费者停止工作如何排查

排查思路与解决方案 #

• 优先排查DoTask未捕获异常导致消费者协程静默退出：
  你当前的消费者逻辑没有异常捕获机制，一旦DoTask在高负载下触发任何未处理异常，整个消费者协程会直接终止，没有任何报错输出，这是此类问题最高发的原因。
  可按以下示例优化消费者代码，补充异常捕获与调试日志：

  import traceback
  
  async def Consumer(client, queue: asyncio.Queue):
      while True:
          try:
              item = await queue.get()
              OutputText(f'Consume task start, queue remaining size: {queue.qsize()}')
              await DoTask(client, item) 
              await asyncio.sleep(timedelay)
              queue.task_done()
              OutputText('Consume task finish')
          except Exception as e:
              OutputText(f'Consumer error: {str(e)}\n{traceback.format_exc()}')
              # 异常场景也要调用task_done，避免队列逻辑异常
              queue.task_done()
  

• 排查事件循环阻塞问题：
  如果DoTask中存在同步阻塞逻辑（比如同步IO调用、CPU密集计算），高CPU/内存负载下会直接卡住整个asyncio事件循环，导致消费者协程得不到调度。可以给DoTask的入口和出口加耗时统计日志，确认是否存在执行时间过长的情况，阻塞类逻辑需要迁移到线程池/进程池中执行。
• 验证队列与协程存活状态：
  asyncio.Queue是进程内内存队列，不存在"连接异常"的问题，你之前的猜测不准确。可以新增一个独立的监控协程，定期打印队列长度、消费者协程的存活状态，确认消费者是否真的处于运行状态，队列是否存在持续堆积。
• 排查运行环境相关问题：
  1. 检查timedelay参数是否存在异常，比如高负载下变成超大值，或者系统时间跳变导致asyncio.sleep无限延后；
  2. Python 3.7的asyncio存在多个高负载下调度挂死的已知bug，可尝试升级到Python 3.8+，或者替换事件循环为uvloop验证问题是否复现；
  3. 高CPU/交换内存占用时，确认是否是你的Python进程占用了绝大多数资源，如果是其他进程抢占资源导致Python进程得不到调度，需要调整系统资源分配策略。

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