如何使用Confluent Python从Kafka指定分区消费数据？

你的代码核心问题是同时调用了subscribe()和assign()方法，这两个方法是互斥的：

• subscribe()会让消费者加入消费者组，触发Kafka的自动分区分配逻辑，直接覆盖你通过assign()手动指定的分区配置，导致仍然会消费其他分区。

正确的做法是只使用assign()手动指定目标分区，不要调用subscribe()，代码示例如下：

from kafka import KafkaConsumer, TopicPartition

# 初始化消费者，无需调用subscribe
consumer = KafkaConsumer(bootstrap_servers='你的Kafka集群地址')

# 明确指定要消费的分区：目标主题的1和2分区
target_partitions = [
    TopicPartition("你的主题名称", 1),
    TopicPartition("你的主题名称", 2)
]
consumer.assign(target_partitions)

# 开始消费指定分区的消息
for msg in consumer:
    print(f"从分区{msg.partition}收到消息: {msg.value}")

额外注意事项：

• 使用assign()时，消费者不属于任何消费者组，需要自行处理消息位移的提交（如果需要持久化消费进度）。
• 如果之前已经调用过subscribe()，建议重新初始化消费者实例，或者先调用consumer.unsubscribe()再执行assign()。

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