使用asyncio调用Confluent Kafka Python阻塞消费者无消息问题咨询

Confluent Kafka Python库的Consumer.poll()是同步阻塞方法，直接在asyncio协程里调用会把整个事件循环卡死——因为asyncio是单线程调度，阻塞调用会霸占CPU，不让出时间片给其他协程，甚至连Kafka消息的接收逻辑都没法正常被调度，所以你收不到消息。

而aiokafka是专门为asyncio设计的异步Kafka客户端，所有IO操作都是非阻塞的，能和asyncio的事件循环完美配合，所以能正常工作。

解决办法 #

方案1：用线程隔离同步消费逻辑 #

把Confluent Kafka的同步消费代码放到单独线程里，和asyncio事件循环解耦，示例代码：

import asyncio
from threading import Thread
from confluent_kafka import Consumer

def sync_kafka_consumer():
    consumer = Consumer({
        'bootstrap.servers': 'server:9092',
        'group.id': 'mygroup',
        'auto.offset.reset': 'earliest',
        'message.max.bytes': 500000,
        'fetch.wait.max.ms': 1000,
        'fetch.min.bytes': 500000,
        'fetch.max.bytes': 500000
    })
    consumer.subscribe(['queuestore'])
    print("Consumer started")
    
    while True:
        msg_dict = consumer.poll(timeout=2000)
        for _, msg in msg_dict.items():
            if msg.error():
                print(f"消费出错: {msg.error()}")
                continue
            # 处理消息
            print(f"收到消息: {msg.value().decode('utf-8')}")

async def main():
    # 启动消费线程，设为守护线程随主程序退出
    consumer_thread = Thread(target=sync_kafka_consumer, daemon=True)
    consumer_thread.start()
    
    # 这里写你的其他异步业务逻辑
    await asyncio.sleep(3600)

if __name__ == "__main__":
    asyncio.run(main())

方案2：直接使用aiokafka #

既然你已经验证过aiokafka能正常工作，继续用它是最省心的选择，它的异步API天生适配asyncio，不需要额外的线程/进程处理。

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