通过JdbcLockRegistry自动缩放组中的Spring Cloud Stream Kinesis Group

Question

我正在尝试创建一个使用Spring云流来消耗来自Kinesis流的事件的消费者应用程序。

我的消费者应用程序在AWS自动缩放组中运行，我希望它可以随时扩展和缩小，而不会影响可以处理流中给定条目的实例数。更具体地说，我希望在任何时间点最多有一个实例处理来自流的事件。

在阅读文档后，我认为可以使用消费者群体实现此功能，Kinesis绑定器甚至在DynamoDB之上提供自己的MetaDataStore和LockRegistry构建。

由于我不想仅仅为了保存锁和元数据而引入DynamoDB，我使用预定义的组使用Jdbc等效实现可以实现相同的结果。

在我的初始测试之后，我看到jdbc存储按预期使用，但是当我增加ASG上的实例时，我注意到消息被消耗了两次（每个实例一次），而不是我在阅读文档后所期望的。

我的房产是：

spring.cloud.stream.bindings.input.destination = stream spring.cloud.stream.bindings.input.group = group spring.cloud.stream.bindings.input.content-type = application / json

锁定注册表负责人e.t.c的配置：

@Bean
public DefaultLockRepository lockRepository(DataSource dataSource) {
    return new DefaultLockRepository(dataSource);
}

@Bean
public LockRegistry lockRegistry(LockRepository lockRepository) {
    return new JdbcLockRegistry(lockRepository);
}

@Bean
public LockRegistryLeaderInitiator leaderInitiator(LockRegistry lockRegistry) {
    LockRegistryLeaderInitiator lockRegistryLeaderInitiator = new LockRegistryLeaderInitiator(lockRegistry);
    lockRegistryLeaderInitiator.setPublishFailedEvents(true);
    return lockRegistryLeaderInitiator;
}

@Bean
public MetadataStore metadataStore(DataSource dataSource) {
    return new JdbcMetadataStore(dataSource);
}

我目前在我的Kinesis上有一个带有一个碎片的流。

我尝试添加更多shard（最多4个），我尝试通过应用程序属性配置并发。

我可以看到元数据存储在我的数据库中按预期填充，并且消费者正在按预期保持状态，但我也看到int_lock表上的多个条目，我没想到会看到（即我希望只看到一个客户注册，但我看到多个）。

所以我的问题是：

1. 有没有人使用jdbc注册表配置使用者？
2. 我的设置有问题吗？
3. 由于文档不是很清楚，锁注册表是否锁定了对Kinesis流的访问权限？

我很确定我在概念方面遗漏了一些东西，但我没有在文档中找到任何东西，the closest to my issue was this issue reported on github但是这让我更加困惑于预期的配置。还不清楚instanceIndex和instanceCount应该如何在ASG环境中工作

Answer 1

我没有看到您的配置有任何问题。 KinesisMessageChannelBinder真的期待一个ConcurrentMetadataStore和LockRegistry豆。你使用的实现已经无关紧要了。

你不需要LockRegistryLeaderInitiator。它没有在活页夹中使用，可能会为您做一些额外的意外工作。

您需要确定，DataSource是所有实例之间的共享RDBMS。此外，您需要确保group选项对于所有实例都是相同的。

锁定逻辑实际上也是基于一个键，如：

this.consumerGroup + ":" + stream + ":" + shardId

所以，没有想法你的配置可能有什么问题......