kafka-python同一Topic并发生产消费时消费者无返回问题如何解决

• 1. 生产者消息未实际发送到Broker：KafkaProducer.send() 是异步方法，仅将消息写入本地缓冲区就返回，默认需要等缓冲区满、定时触发或实例销毁时才会批量发送到Broker。你顺序执行时生产者函数跑完后实例销毁自动触发flush，所以消费者能读到数据；但多进程并发运行时生产者进程一直处于循环中，未触发自动flush逻辑，消息全部堆积在本地缓冲区，Broker没有数据消费者自然读不到。
• 2. 多进程fork导致客户端状态损坏：你在主进程提前初始化了KafkaProducer和KafkaConsumer实例，使用multiprocessingfork子进程时，客户端内部维护的网络连接、元数据、缓冲区状态会被破坏，子进程拿到的实例无法正常和Broker交互。
• 3. （可选影响项）auto_offset_reset='latest'配置：如果消费者启动时间早于第一条消息到达Broker的时间，该配置会让消费者从最新的偏移量开始消费，不会丢失消息，但如果消费者启动前已经有消息存在，就会跳过历史消息，可根据需求调整为earliest。

不需要引入额外组件，仅需调整代码逻辑即可实现预期的并发生产消费效果，调整点如下：

1. 将生产者、消费者的初始化逻辑移动到对应子进程的执行函数内部，避免主进程fork带来的状态异常
2. 生产者每次调用send()后追加调用flush()，强制把消息立刻发送到Broker

import time
from kafka import KafkaProducer, KafkaConsumer
import multiprocessing

TOPIC = 'fortest'

def store_message():
    # 生产者初始化放到当前进程内部
    producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
    for _ in range(100):
        msg = b'message'
        producer.send(topic=TOPIC, value=msg)
        # 新增flush调用，强制发送消息到Broker
        producer.flush()
        print(f'{msg} sent by Producer')
        time.sleep(3)

def get_processed_message():
    # 消费者初始化放到当前进程内部
    consumer = KafkaConsumer(
        TOPIC,
        bootstrap_servers=['localhost:9092'],
        # 如果需要消费启动前的历史消息，可将下面参数改为'earliest'
        auto_offset_reset='latest',
        group_id='my-consumer-1'
    )
    while True:
        messages = consumer.poll(timeout_ms=5000)
        if not messages:
             print('wait for messsages')
             time.sleep(5)
        else:
            print(f"Get messages: {messages.values()}")

if __name__ == '__main__':
    produce_initial_message = multiprocessing.Process(target=store_message)
    consume_processed_message = multiprocessing.Process(target=get_processed_message)
    produce_initial_message.start()
    consume_processed_message.start()

如果是拆分为两个独立脚本分别运行生产者和消费者，同样只需要在生产者的send方法后加producer.flush()即可，不需要其他调整。

内容的提问来源于stack exchange，提问作者Назарій Кушнір