Kafka消费者使用getmany处理大量消息时停滞问题求助

我使用AIOKafkaConsumer的getmany方法消费消息，当需要处理总计650条消息（预计耗时约3天）时，仅完成100-150条的处理（耗时12至24小时）后便不再继续处理，但消费者连接并未关闭，新增消息仍可正常处理。测试25条消息（3小时全部完成）、100条消息（12小时全部完成）的场景时均无消息跳过，仅在消息量较大（650条）时出现该问题。

async def consume(loop,lock):
    logger.info('Inside Consume')

    consumer = AIOKafkaConsumer(KAFKA_TOPIC,
                        loop=loop,
                        bootstrap_servers=bootstrap_servers,
                        group_id=group_id,           
                        enable_auto_commit=enable_auto_commit,       
                        auto_commit_interval_ms=auto_commit_interval_ms,  
                        auto_offset_reset=auto_offset_reset,  
                        max_poll_records= 1,
                        max_poll_interval_ms=1500000,
                        rebalance_timeout_ms=1500000)
              
    await consumer.start()
    while True:
        result = await consumer.getmany(timeout_ms=1500000, max_records=1)
        for tp, msg in result.items(): 
            if msg:
                message = msg[0].value.decode()
                message_json = json.loads(message)
                data_encoded = message_json['payload']

                if data_encoded.get('content') is not None:
                    msg = data_encoded['content']
                    message_dict_dum = base64.b64decode(msg)
                    message_dict = json.loads(message_dict_dum)
                    
                    if message_dict.get("key1") is not None:
                        if message_dict["key1"] == "something":
                            logger.info('Hurray new message to process')
                            logger.info(message_dict["id"])
                            await process(data_encoded['content'],lock)
                            

async def process(msg, lock):
    async with lock:
        logger.info('processing... ->{}'.format(msg))
        await processmessage(msg)
        logger.info('done processing')        


if __name__ == "__main__":
    logger.info('Inside main')

    loop = asyncio.new_event_loop()
    lock = asyncio.Lock(loop=loop)

    loop.create_task(consume(loop, lock))
    loop.run_forever()

• 事件循环阻塞：processmessage如果包含同步阻塞操作（如同步数据库调用、耗时的CPU计算），会卡住整个异步事件循环，导致无法执行getmany拉取后续消息。建议将同步逻辑放到线程池执行，避免阻塞事件循环。
• 偏移量提交异常：启用自动提交时，偏移量提交时机依赖于poll间隔，若处理过程中出现延迟，可能导致偏移量提交不及时，后续拉取时出现偏移混乱。改为手动提交偏移量，确保消息处理完成后再提交。
• 分区拉取卡点：可以在处理消息时打印当前分区和偏移量，排查是否卡在某个特定偏移点，确认是否存在消息损坏或拉取异常。
• 重平衡异常：长时间运行后可能触发消费者组重平衡，若分区分配异常会导致无法拉取消息。添加重平衡回调函数，监听分区分配/回收事件，排查异常。

1. 手动提交偏移量

# 修改consumer初始化参数
consumer = AIOKafkaConsumer(KAFKA_TOPIC,
                            # ... 其他参数不变
                            enable_auto_commit=False)

# 处理完消息后添加提交逻辑
await process(data_encoded['content'],lock)
# 提交当前消息的下一个偏移量
await consumer.commit({tp: msg[0].offset + 1})

2. 避免事件循环阻塞
   如果processmessage是同步函数，改为异步执行：

async def processmessage(msg):
    loop = asyncio.get_running_loop()
    # 将同步逻辑放到线程池执行
    await loop.run_in_executor(None, sync_processmessage, msg)

3. 添加调试日志
   在处理消息时打印分区和偏移量，方便排查卡点：

logger.info(f"处理分区 {tp.partition} 偏移量 {msg[0].offset} 的消息")

4. 监听重平衡事件

def on_rebalance(consumer, partitions):
    logger.info(f"触发重平衡，当前分区: {[p.partition for p in partitions]}")

consumer.subscribe([KAFKA_TOPIC], on_rebalance=on_rebalance)

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