弹簧集成 Kinesis 适配器和消费者组

Question

我有一个使用1.1.0.RELEASE版本开发的 kinesis 消费者应用程序spring-integration-aws。

在我的测试中，我在同一使用者组中运行此应用程序的两个实例，并从具有两个分片的流中使用。在我的测试中，我意识到KinesisMessageDrivenChannelAdapter会以三种方式分发消息：

1. 传递到一个使用者的所有消息
2. 分发给两个使用者的消息(不均匀(
3. 两个使用者都收到相同的消息

从生产者端来看，消息在两个分片之间均匀分布。我想知道 kinesis 适配器如何在消费者之间分发消息，如果支持，如何在消费者之间均匀分布。

谢谢

更新(适配器配置(

@Bean
public KinesisMessageDrivenChannelAdapter kinesisInboundChannelAdapter(
AmazonKinesis amazonKinesis) {
String[] streamNames = this.consumerClientProperties.getKinesis().getStreamNames();
KinesisMessageDrivenChannelAdapter adapter =
new KinesisMessageDrivenChannelAdapter(amazonKinesis, streamNames);
adapter.setConverter(null);
adapter.setOutputChannel(new QueueChannel());
adapter.setCheckpointStore(dynamoDbMetaDataStore());
adapter.setCheckpointMode(CheckpointMode.record);
adapter.setStartTimeout(10000);
adapter.setConsumerGroup(consumerClientProperties.getName());
adapter.setListenerMode(ListenerMode.record);
adapter.setDescribeStreamRetries(1);
return adapter;
}
@Bean
public DynamoDbMetadataStore dynamoDbMetaDataStore() {
DynamoDbMetadataStore dynamoDbMetaDataStore = new DynamoDbMetadataStore(amazonDynamoDB(),
consumerClientProperties.getName());
return dynamoDbMetaDataStore;
}

Answer 1

建议大家升级到最新的 Spring Integration AWS2.0： https://spring.io/blog/2018/08/21/spring-integration-for-aws-2-0-ga-and-spring-cloud-stream-kinesis-binder-1-0-ga

在 Kinesis 消费者级别上进行了大量修复，现在我们有一个领导者选举，不要多次订阅同一个分片。

这个想法是在处理记录时有严格的排序，因此每个集群只有一个线程可以访问一个分片。不过，该线程可能会处理多个分片。

无论如何，如果您使用应用程序的两个实例，则需要注入基于共享数据的MetadataStore，例如DynamoDbMetadataStore.