基于Kafka的大模型数据流处理实践

最近在为一个大模型推理服务设计数据流处理架构时，踩了一个大坑。一开始我们直接在应用层用Python的asyncio处理消息队列，结果在高峰期直接把服务搞崩了。

问题复现

# 错误做法
from kafka import KafkaConsumer
import asyncio

class BadProcessor:
    def __init__(self):
        self.consumer = KafkaConsumer(
            'model-input',
            bootstrap_servers='localhost:9092',
            value_deserializer=lambda x: json.loads(x)
        )
    
    async def process_messages(self):
        for msg in self.consumer:
            # 直接处理，没有限流
            result = await self.model_predict(msg.value)
            # 同步写回

正确实践

采用Kafka + 消费者组 + 限流机制的架构：

# 正确做法
from kafka import KafkaConsumer
from asyncio import Semaphore
import asyncio


class GoodProcessor:
    def __init__(self, max_concurrent=10):
        self.consumer = KafkaConsumer(
            'model-input',
            bootstrap_servers='localhost:9092',
            group_id='model-group',
            value_deserializer=lambda x: json.loads(x)
        )
        self.semaphore = Semaphore(max_concurrent)
        
    async def process_message(self, msg):
        async with self.semaphore:
            result = await self.model_predict(msg.value)
            return result

    async def run(self):
        tasks = []
        for msg in self.consumer:
            task = asyncio.create_task(self.process_message(msg))
            tasks.append(task)
            
            # 控制并发数
            if len(tasks) >= 100:
                await asyncio.gather(*tasks)
                tasks = []

核心经验

消费者组配置要合理，避免重复消费
添加限流机制防止模型过载
使用异步处理提高吞吐量

DryFire · 2026-01-08T10:24:58

别再用asyncio直接消费Kafka了，我亲眼看着服务在高峰期直接挂掉，根本没得救。这种做法就是把消费者和生产者绑死，一旦模型推理慢一点，整个链路就堵死了。

心灵画师 · 2026-01-08T10:24:58

正确的姿势是必须加消费者组+限流机制，不然就是拿服务器性能做赌注。别觉得Kafka是万能的，它只是个消息中间件，处理逻辑还得靠你自己的架构设计。

Will241 · 2026-01-08T10:24:58

我建议把并发数控制在10以内，别贪多。大模型推理本身就很吃资源，再加asyncio的调度开销，很容易把CPU打满。用信号量限制并发才是王道。

NiceWolf · 2026-01-08T10:24:58

别再想着自己写消费者了，直接上成熟的流处理框架比如Flink或者Spark Streaming。Kafka只是个载体，真正的处理逻辑必须有容错和限流能力，否则就是给故障埋雷。

基于Kafka的大模型数据流处理实践