1

AFAIK，

kafka 中引入了分区和（消费者）组的概念来实现并行。我正在通过python使用kafka。我有一个特定的主题，它有（比如说）2 个分区。这意味着，如果我启动一个包含 2 个消费者的消费者组，他们将被映射（订阅）到不同的分区。

但是，kafka在 python 中使用库时，我遇到了一个奇怪的问题。我启动了 2 个具有基本相同组 ID 的消费者，并启动了线程让他们消费消息。

但是，kafka-stream 中的每条消息都被他们俩消费了！！这对我来说似乎很荒谬，甚至在概念上也不正确。无论如何我可以手动将消费者映射到某些（不同的）分区（如果它们没有自动映射到不同的分区）？

这是代码：

from kafka import KafkaConsumer
import thread

def con1(consumer):
    for msg in consumer:
        print msg

consumer1 = KafkaConsumer('k-test', group_id='grp1', bootstrap_servers=['10.50.23.120:9092'])
consumer2 = KafkaConsumer('k-test', group_id='grp1', bootstrap_servers=['10.50.23.120:9092'])

thread.start_new_thread(con1, (consumer1,))
thread.start_new_thread(con1, (consumer2,))

这是我使用 kafka-console-producer 生成的一些消息的输出：

ConsumerRecord(topic=u'k-test', partition=0, offset=47, timestamp=None, timestamp_type=None, key=None, value='polki')
ConsumerRecord(topic=u'k-test', partition=0, offset=47, timestamp=None, timestamp_type=None, key=None, value='polki')
ConsumerRecord(topic=u'k-test', partition=0, offset=48, timestamp=None, timestamp_type=None, key=None, value='qwewrg')
ConsumerRecord(topic=u'k-test', partition=0, offset=48, timestamp=None, timestamp_type=None, key=None, value='qwewrg')
ConsumerRecord(topic=u'k-test', partition=0, offset=49, timestamp=None, timestamp_type=None, key=None, value='shgjas')
ConsumerRecord(topic=u'k-test', partition=0, offset=49, timestamp=None, timestamp_type=None, key=None, value='shgjas')

而预期是其中之一。顺便说一句，这个主题k-test有 2 个分区。

4

4 回答 4

0

我猜您正在使用 Kafka 0.8 或更低版本，根据文档不支持此功能：

...但是，某些功能将仅在较新的经纪人上启用；例如，完全协调的消费者组——即，对同一组中的多个消费者进行动态分区分配——需要使用 0.9+ 个 kafka 代理......

于 2016-12-21T00:56:56.443 回答
0

尝试运行 bin/kafka-consumer-groups.sh 命令行工具来验证您使用的 Python Kafka 客户端是否支持正确的消费者组管理。如果两个消费者确实在同一个组中，那么他们应该从互斥分区获取消息。

于 2017-06-06T05:06:38.950 回答
0

根据我的经验，密钥必须长于 4 个字符，否则一切都会转到 partition0

于 2021-09-01T11:17:09.430 回答
0 
from kafka import KafkaConsumer
from kafka import TopicPartition

TOPIC = "k-test"
PARTITION_0 = 0
PARTITION_1 = 1

consumer_0 = KafkaConsumer(
    TOPIC, group_id='grp1', bootstrap_servers=['10.50.23.120:9092']
)
consumer_1 = KafkaConsumer(
    TOPIC, group_id='grp1', bootstrap_servers=['10.50.23.120:9092']
)
topic_partition_0 = TopicPartition(TOPIC, PARTITION_0)
topic_partition_1 = TopicPartition(TOPIC, PARTITION_1)
# format: topic, partition
consumer_0.assign([topic_partition_0])
consumer_1.assign([topic_partition_1])

assign() 可能对你有用，但是一旦你使用它，当消费者停止工作时，kafka 将不会自动平衡消费者。

于 2017-06-06T03:47:34.843 回答