AIOKafka在多进程类中异常退出，无法消费消息问题排查

核心问题分析 #

1. 序列化方法参数不匹配：_deserializer被定义为类的普通方法，但未添加@staticmethod装饰器。AIOKafkaConsumer调用该方法时，会默认传入实例self作为第一个参数，而方法本身仅接受serialized参数，触发TypeError异常，导致消费流程提前终止，consumer被停止后进程直接退出。
2. 任务等待逻辑顺序错误：原代码在停止consumer后才等待异步任务完成，可能导致消息处理任务未执行完毕就被终止，同时异常会直接触发finally块的consumer停止操作，加速进程退出。

修正方案 #

1. 修复序列化方法 #

给_deserializer添加@staticmethod装饰器，使其成为类的静态方法，避免参数不匹配问题。

2. 调整任务等待逻辑 #

将异步任务的等待操作移到try块内部、consumer停止之前，确保所有消息处理任务执行完毕后再关闭consumer。

3. 添加异常捕获（可选） #

在消费逻辑中添加异常捕获，便于排查后续可能出现的问题。

修正后的完整代码 #

Consumer类代码 #

import json
import logging
import asyncio
from multiprocessing import Process
from aiokafka import AIOKafkaConsumer

# 假设process_msg是已实现的异步消息处理函数
async def process_msg(msg, delay):
    logging.info(f"Processing message content: {msg.value}")
    await asyncio.sleep(delay)

class Consumer(Process):
    def __init__(self, topic, **kwargs):
        self.topic = topic
        super().__init__(**kwargs)

    @staticmethod
    def _deserializer(serialized):
        return json.loads(serialized)

    async def _consume(self):
        consumer = AIOKafkaConsumer(
            self.topic,
            group_id="Deployment",
            value_deserializer=self._deserializer,
            bootstrap_servers='localhost:30322',
        )

        await consumer.start()
        tasks = []
        try:
            async for msg in consumer:
                logging.info("***** reading message *****")
                tasks.append(asyncio.create_task(process_msg(msg, 1)))
            # 等待所有已提交的消息处理任务完成
            await asyncio.gather(*tasks)
        except Exception as e:
            logging.error(f"Consumption failed with error: {str(e)}", exc_info=True)
        finally:
            await consumer.stop()

    def run(self):
        asyncio.run(self._consume())

主启动代码 #

import logging

# 配置日志级别为INFO，确保能捕获关键信息
logging.basicConfig(level=logging.INFO)

num_procs = 1
processes = [Consumer("deployment_requests") for _ in range(num_procs)]

for p in processes:
    p.start()

for p in processes:
    logging.info(f'pid is {p.pid}')

for p in processes:
    p.join()
    logging.info(f'pid is {p.pid}')

验证说明 #

修正后，消费者进程会正常加入Kafka消费组，进入消息循环等待消息，收到消息后会打印***** reading message *****日志，并执行消息处理任务。任务执行完毕后，若没有新消息，进程会持续等待，不会主动退出。

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