Kafka Consumer与Producer能否同线程运行？同线程消费无消息排查

这个问题和Producer、Consumer运行在同一线程完全无关，Kafka从设计上就支持同线程运行生产者和消费者实例，你遇到的问题是消费者默认配置+代码时序错误共同导致的，核心原因有两个：

• 消费者默认的偏移量重置策略是latest
• 你在消费者还没完成消费组入组、分区分配流程时就把所有消息发完了

具体原理说明 #

1. auto.offset.reset配置的默认行为
   confluent-kafka-python客户端的消费者默认配置中，auto.offset.reset值为latest：当消费组没有已提交的有效偏移量时，消费者会从自身完成分区分配、成功加入消费组的时刻对应的最新消息偏移量开始消费，这个时刻之前产生的所有消息都不会被拉取。
2. 消费者入组流程的触发时机
   调用consumer.subscribe()仅仅是在客户端本地标记了要订阅的主题，并不会立刻和Broker完成消费组协调、重平衡、分区分配流程——所有和消费组相关的心跳、重平衡、分区分配、偏移量获取逻辑，只有在你调用consumer.poll()的时候才会推进。
3. 你的代码时序问题
   你的代码执行顺序是：
   • 创建消费者、调用subscribe（仅本地标记订阅，未实际入组）
   • 立刻创建生产者，发送完所有101条消息，flush()等待生产者确认消息发送成功
   • 才开始循环调用poll()尝试拉取消息
     当你第一次调用poll()时，消费者才开始触发重平衡流程，等它成功分配到分区、确定消费起始偏移量时，你之前发的所有消息都已经存在于Broker中，这时候latest策略下的起始偏移量正好是所有已发送消息之后的位置，自然拉不到任何之前的消息。
     你加的sleep(10)逻辑完全无效，因为sleep期间你没有调用poll()，消费者的入组流程根本没有推进，睡再久也没用。
4. 为什么独立运行的消费者能收到消息
   并行运行的独立消费者是提前启动的，早就通过持续调用poll()完成了入组和分区分配，确定了消费起始位置，你发送消息时它已经处于正常消费状态，所以能收到所有消息。

你只需要调整代码逻辑+按需补充配置即可，不需要拆分线程：

• 调整执行顺序：订阅主题后，先调用一段时间poll()，确保消费者完成重平衡、拿到分区分配之后，再启动生产者发消息
• 如果你需要消费主题内的历史消息，可以在消费者配置中添加"auto.offset.reset": "earliest"，注意该配置仅在消费组无有效已提交偏移量时生效
• （可选）可以添加分区分配回调，明确感知到分区分配完成后再开始发消息，避免时序不确定问题

修复后的参考代码 #

from confluent_kafka import Producer, Consumer

IP = "localhost:9092"
TRIES = 100
topic = "topic1"
group = "group1"

# 补充auto.offset.reset配置，按需选择earliest/latest
consumer_config = {
    "bootstrap.servers": IP,
    "group.id": group,
    "auto.offset.reset": "earliest"
}
producer_config = {"bootstrap.servers": IP}

def confirm(err, msg):
    if err:
        print(err)
    else:
        print(f"delivered msg to {msg.topic()} [{msg.partition()}] offset {msg.offset()}")

consumer = Consumer(consumer_config)
producer = Producer(producer_config)
consumer.subscribe([topic])

# 先运行poll，等待消费者完成入组和分区分配，最多等10秒
wait_rebalance_counter = 0
while wait_rebalance_counter < 10:
    consumer.poll(timeout=1)
    wait_rebalance_counter += 1

# 确认消费者入组完成后再发消息
producer.produce(value="init", topic=topic, on_delivery=confirm)
for i in range(1, TRIES + 1):
    producer.produce(value=str(i), topic=topic)
# 等待所有消息发送确认
producer.flush()

messages = []
counter = 0
while counter < 10:
    message = consumer.poll(timeout=1)
    if message and not message.error():
        counter = 0
        messages.append(message.value().decode('utf-8'))
    else:
        counter += 1
print(messages)
consumer.close()

Kafka客户端（包括librdkafka底层的confluent-kafka-python）本身的网络IO、消息缓存等逻辑都是在独立的后台线程运行的，业务线程中同线程创建、调用Producer和Consumer没有任何限制，不存在不支持同线程运行的说法。

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