基于aiokafka的多消费-处理-多生产架构性能瓶颈排查优化

问题定位 #

原代码处理速度仅约50条/秒，核心瓶颈集中在以下几点：

1. 消费者拉取效率低：默认参数下每次拉取消息量极小，频繁与Kafka Broker交互，且未开启批量拉取；若输入Topic分区数不足4，部分消费者会空闲。
2. 处理器串行阻塞：仅单个处理任务，完全串行处理消息；process_message中的print是同步阻塞操作，严重拖慢异步事件循环。
3. 生产者（注释部分）性能差：使用send_and_wait同步等待消息确认，每条消息单独发送，无批量配置，发送效率极低。

优化方案 #

一、消费者端优化 #

• 配置批量拉取参数：调整fetch_min_bytes（拉取最小字节数）和fetch_max_wait_ms（最长等待时间），让消费者攒够足够数据再拉取，减少Broker交互次数。
• 匹配分区与消费者数量：确保输入Topic的分区数≥消费者数量（4个），避免消费者空闲。
• 批量入队：攒一批消息再放入异步队列，减少queue.put的频繁调用开销。
• 优化序列化/反序列化：使用更高效的JSON库（如ujson）替代标准库json，提升解析速度。

二、处理器端优化 #

• 多任务并行处理：启动多个处理任务，并行消费队列中的消息，打破单线程瓶颈。
• 移除同步阻塞操作：删除调试用的print，或替换为异步日志框架（如structlog），避免阻塞事件循环。
• 可选：批量处理：若业务允许，攒一批消息再统一处理，进一步减少上下文切换开销。

三、生产者端优化 #

• 使用异步发送：用producer.send()替代send_and_wait()，无需等待Broker确认，大幅提升发送吞吐量。
• 配置批量发送参数：设置linger_ms（等待攒批时长）和batch_size（单批最大字节数），让生产者自动攒批发送。
• 多生产者实例：根据输出Topic分区数调整生产者数量，提升发送并行度。

优化后完整代码 #

import asyncio
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
import ujson  # 需提前安装：pip install ujson

BROKERS = [
    "BROKER0:PORT",
    "BROKER1:PORT",
    "BROKER2:PORT",
]

GROUP_ID = "group_id"
TOPIC_INPUT = "topic_input"
TOPIC_OUTPUT = "topic_output"
# 批量配置
CONSUME_BATCH_SIZE = 100  # 每批攒100条消息
CONSUME_FETCH_MIN_BYTES = 1 * 1024 * 1024  # 1MB
CONSUME_FETCH_MAX_WAIT_MS = 500  # 最长等待500ms
PRODUCE_BATCH_SIZE = 16384  # 16KB
PRODUCE_LINGER_MS = 10  # 等待10ms攒批


async def consume(queue):
    consumer = AIOKafkaConsumer(
        TOPIC_INPUT,
        bootstrap_servers=BROKERS,
        value_deserializer=lambda m: ujson.loads(m.decode('utf-8')),
        group_id=GROUP_ID,
        auto_offset_reset="latest",
        # 批量拉取配置
        fetch_min_bytes=CONSUME_FETCH_MIN_BYTES,
        fetch_max_wait_ms=CONSUME_FETCH_MAX_WAIT_MS,
    )
    await consumer.start()
    try:
        batch = []
        async for message in consumer:
            processed_message = {
                "timestamp": message.timestamp,
                "col1": message.value["col1"],
                "col2": message.value["col2"],
                "col3": message.value["col3"],
            }
            batch.append(processed_message)
            # 攒够批量或达到阈值时入队
            if len(batch) >= CONSUME_BATCH_SIZE:
                await queue.put(batch)
                batch = []
        # 处理剩余的消息
        if batch:
            await queue.put(batch)
    finally:
        await consumer.stop()


async def process_message(message):
    # 替换为异步日志（若需记录）
    # logger.info("Processing message", message=message)
    return message


async def process_messages(queue, output_queue):
    while True:
        # 批量取出消息处理
        batch = await queue.get()
        processed_batch = [await process_message(msg) for msg in batch]
        await output_queue.put(processed_batch)
        queue.task_done()


async def produce(output_queue):
    producer = AIOKafkaProducer(
        bootstrap_servers=BROKERS,
        value_serializer=lambda m: ujson.dumps(m).encode('utf-8'),
        # 批量发送配置
        batch_size=PRODUCE_BATCH_SIZE,
        linger_ms=PRODUCE_LINGER_MS,
    )
    await producer.start()
    try:
        while True:
            batch = await output_queue.get()
            # 批量发送消息，无需等待确认
            for msg in batch:
                await producer.send(TOPIC_OUTPUT, msg)
            output_queue.task_done()
    finally:
        await producer.stop()


async def main():
    # 根据机器内存调整队列大小
    queue = asyncio.Queue(maxsize=1000)
    output_queue = asyncio.Queue(maxsize=1000)

    # 消费者数量匹配输入Topic分区数
    consumers = [asyncio.create_task(consume(queue)) for _ in range(4)]
    # 处理器任务数可根据CPU核心数调整
    process_tasks = [asyncio.create_task(process_messages(queue, output_queue)) for _ in range(4)]
    # 生产者数量匹配输出Topic分区数
    producers = [asyncio.create_task(produce(output_queue)) for _ in range(3)]

    await asyncio.gather(*consumers, *process_tasks, *producers)


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

关键注意事项 #

• 队列大小调整：避免设置过大导致内存溢出，建议根据机器内存和消息平均大小计算合理值。
• 分区数匹配：消费者/生产者数量不要超过对应Topic的分区数，否则多余的实例会空闲。
• 监控积压：通过queue.qsize()监控队列积压情况，动态调整消费者、处理器、生产者的数量配比。
• 依赖优化：确保aiokafka和ujson为最新版本，避免已知性能问题。

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