Spring Cloud Stream Kinesis绑定器-并发

Question

我用以下组件构建了一个spring boot kinesis消费者：

• Spring boot（版本-2.1.2.Release)
• Spring cloud（version-greenwich.release)
• Spring cloud stream kinesis绑定器（版本-1.1.0.发行版）

假设我有3个消费者实例部署到PCF（通过在MANIFEST.YML文件中将instances属性设置为3，这在cf推送期间使用）。

所有3个实例都具有以下属性：

Spring.cloud.stream.Bindings..Consumer.concurrency=5

spring.cloud.stream.bindings..group=my-consumer-group

spring.cloud.stream.kinesis.binder.checkpoint.table=my-metadata-dynamodb-table

spring.cloud.stream.kinesis.binder.locks.table=my-locks-dynamodb-table

假设这些事件是由制片人按以下顺序发送到kinesis的

event5（流中最近的事件）-event4-event3-event2-event1（流中的第一个事件）

null

Answer 1

请参阅KinesisMessageDrivenChannelAdapter中的Concurrency选项JavaDocs:

/**
 * The maximum number of concurrent {@link ConsumerInvoker}s running.
 * The {@link ShardConsumer}s are evenly distributed between {@link ConsumerInvoker}s.
 * Messages from within the same shard will be processed sequentially.
 * In other words each shard is tied with the particular thread.
 * By default the concurrency is unlimited and shard
 * is processed in the {@link #consumerExecutor} directly.
 * @param concurrency the concurrency maximum number
 */
public void setConcurrency(int concurrency) {

因此，由于在一个流中只有一个碎片，因此只有一个活动线程在这个碎片上迭代ShardIterator。

关键是我们必须在一个线程中处理来自单个碎片的记录。这样，我们保证了一个正确的顺序，加上检查点是为最高的序号。

请更多地调查什么是AWS Kinesis以及它是如何工作的。