KafkaConsumer阻塞其他asyncio.Task正常运行的原因及解决方法

1. 你使用的kafka-python库提供的KafkaConsumer是同步阻塞实现，for msg in self.consumer迭代过程会一直占用当前线程等待Kafka消息，不会主动让出asyncio事件循环的控制权，导致事件循环被堵死，ping_task等其他异步任务完全没有被调度执行的机会。
2. 额外隐患：你在on_ready方法中创建的两个asyncio.Task是局部变量，方法执行结束后变量引用被销毁，任务有被Python垃圾回收机制意外终止的风险。

方案1（推荐）：替换为异步Kafka客户端 #

改用专门适配asyncio生态的aiokafka库，消费逻辑全程异步非阻塞，不会卡住事件循环。
首先安装依赖：

pip install aiokafka

修改后的代码示例：

import asyncio
import json
import discord
from aiokafka import AIOKafkaConsumer


class DiscordBot(discord.Client):
    def __init__(self):
        super().__init__()
        # 初始化异步consumer，先不启动
        self.consumer = AIOKafkaConsumer(
            "my_topic",
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            bootstrap_servers="127.0.0.1:9092" # 替换为你的kafka地址
        )
        
    async def on_ready(self):
        # 把task存为实例属性，避免被GC回收
        self.ping_task = asyncio.create_task(self.ping())
        self.echo_kafka_msgs_task = asyncio.create_task(self.echo_kafka_msgs())
        
    async def echo_kafka_msgs(self):
        # 先启动consumer
        await self.consumer.start()
        try:
            while True:
                self._logger.info("Waiting for new messages...")
                # 异步迭代，没有消息时自动让出事件循环
                async for msg in self.consumer:
                    print(msg.value)
        finally:
            # 退出时关闭consumer
            await self.consumer.stop()
            
    async def ping(self):
        i = 0
        while True:
            print("i=", i)
            i += 1
            await asyncio.sleep(1)    

if __name__ == '__main__':
    client = DiscordBot()
    client.run("my_token")

方案2：同步Kafka客户端线程池隔离 #

如果你必须保留现有kafka-python的使用，可以把阻塞的消费逻辑放到单独线程池中执行，避免阻塞事件循环，仅需修改echo_kafka_msgs方法即可：

async def echo_kafka_msgs(self):
    loop = asyncio.get_running_loop()
    while True:
        self._logger.info("Waiting for new messages...")
        # 把同步拉取消息的逻辑扔到线程池运行，异步等待结果
        msg = await loop.run_in_executor(None, next, self.consumer)
        print(msg.value)

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