Dav*_*iro 5 java apache-kafka kafka-consumer-api
我有一个主题,一个分区和两个构成一个消费者组的消费者进程.
这样,消息始终传递给单个消费者.StickyAssignor用于优先选择已经分配给重新平衡分区的消费者.
我一直在玩这个设置,并发现在某些情况下,消息被传递给两个消费者,这打破了消费者群体的目的.
方案如下:
使用RangeAssignor/RoundRobinAssignor时,不会发生这种情况.
我错过了什么,或者这是卡夫卡的一个错误?
这是我的消费者代码:
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("client.id", consumerId);
props.put("enable.auto.commit", "false");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("partition.assignment.strategy", StickyAssignor.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singleton("my-events"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
Run Code Online (Sandbox Code Playgroud)
归档时间: |
|
查看次数: |
483 次 |
最近记录: |