我已经建立了一个生产者Spring云流应用程序和Kafka作为活页夹。以下是application.yml:
spring:
cloud:
stream:
instanceCount : 1
bindings:
output:
destination: topic-sink
producer:
partitionSelectorClass: com.partition.CustomPartition
partitionCount: 1
...
我有两个示例(同一个应用程序在一个jvm上运行)作为使用者。以下是application.yml:
spring:
cloud:
stream:
bindings:
input:
destination: topic-sink
group: hdfs-sink
consumer:
partitioned: true
...
我对kafka组的理解是,对于同一组中的那些消费者,消息只会被消费一次。假设,如果生产者应用程序生成消息a、b,并且同一组中有两个消费者应用程序,则消费者1将读取消息a,消费者2将读取消息b、c。然而,我的消费者正在消费同样的信息。我的假设错了吗?
1条答案
按热度按时间htzpubme1#
我有办法了,谢谢你。对于1个分区和1个使用者。我在spring cloud stream应用程序中为生产者/消费者共享解决方案。制作人:
spring: cloud: stream: instanceCount : 1 bindings: output: destination: topic-sink producer: partitionSelectorClass: com.partition.CustomPartition partitionCount: 1
消费者:spring: cloud: stream: instanceIndex: 0 #between 0 and instanceCount - 1 instanceCount: 1 bindings: input: destination: topic-sink group: hdfs-sink consumer: partitioned: true kafka: binder: autoAddPartitions: true