使用 Kinesis 客户端库

连接到数据流

枚举数据流中的分片

使用租约与其工作人员协调分片关联

为其管理的每个分片实例化记录处理器

从数据流中提取数据记录

将记录推送到对应的记录处理器

对已处理记录进行检查点操作

当工作线程实例计数发生变化或重新分片数据流时（分片被拆分或合并）时，平衡分片工作人员关联（租赁）

KCL 1.x

有关更多信息，请参阅 。开发 KCL 1.x 使用者

KCL 2.x

有关更多信息，请参阅 。开发 KCL 2.x 使用者

KCL 使用者应用— 使用 KCL 自定义构建并旨在读取和处理数据流中记录的应用程序。

消费者应用实例-KCL 消费者应用程序通常是分布式的，一个或多个应用程序实例同时运行，以便协调故障并动态负载平衡数据记录处理。

工作线程— KCL 消费者应用程序实例用于开始处理数据的高级别类。

工作人员初始化并监督各种任务，包括同步分片和租赁信息、跟踪分片分配以及处理分片中的数据。工作程序为 KCL 提供了使用者应用程序的配置信息，例如将记录此 KCL 使用者应用程序要处理的数据流的名称以及Amazon访问此数据流所需的凭据。工作人员还启动该特定的 KCL 消费者应用程序实例，将数据记录从数据流传送到记录处理器。

租赁— 定义工作器和分片之间绑定的数据。分布式 KCL 消费者应用程序使用租赁将数据记录处理跨工作人员队列进行分区。在任何给定时间，每个数据记录分片都由leaseKey变量。

默认情况下，工作人员可以持有一个或多个租约（取决于maxLeasesFor工作线程同时提交。

例如，如果您有一个具有工作器 A 的消费者应用程序实例 A 正在处理包含 4 个分片的数据流，则工作器 A 可以同时保存分片 1、2、3 和 4 的租约。但是如果你有两个消费者应用程序实例：A 和 B 使用工作器 A 和工作器 B，并且这些实例正在处理包含 4 个分片、工作器 A 和工作器 B 的数据流不能同时持有分片 1 的租约。一名工作人员持有特定分片的租约，直到准备停止处理此分片的数据记录或失败为止。当一名工作人员停止持有租约时，另一名员工接管并持有租约。

有关更多信息（这些是 Java KCL 存储库），请参阅https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/SRC/main/java/com/亚马逊/服务/Kinesis/租赁/impl/租赁.java对于 KCL 1.x 和https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java对于 KCL 2.x。

租赁表-一个独特的 Amazon DynamoDB 表，用于跟踪 KCL 消费者应用程序工作人员租用和处理的 KDS 数据流中的分片。在运行 KCL 使用者应用程序时，租赁表必须与数据流中的最新分片信息保持同步（在工作程序内部和所有工作线程之间）。有关更多信息，请参阅 使用租赁表跟踪 KCL 消费者应用程序处理的碎片。

录制处理器— 定义 KCL 消费者应用程序如何处理从数据流中获取的数据的逻辑。在运行时，KCL 消费者应用程序实例化一个工作器，此工作程序为其持有租约的每个分片实例化一个记录处理器。

检查点：分片的最新检查点序号。此值在数据流中的所有分片中都是唯一的。

checkpointSubSequence数量：使用 Kinesis Producer Library 的聚合功能时，这是一个扩展检查点它跟踪 Kinesis 记录中的单个用户记录。

leaseCounter：用于租赁版本控制，以便工作程序可以检测其租赁已由其他工作程序获取。

leaseKey：租赁的唯一标识符。每个租赁特定于数据流中的一个分片，一次由一个工作程序持有。

leaseOwner：持有此租赁的工作程序。

ownerSwitchesSince检查点：自上次写入检查点以来，此租赁更改了工作程序的次数。

parentShardId：用于确保在开始子分片的处理之前已经完全处理了父分片。这可以确保记录按照放入流中的相同顺序处理。

hashrange：由PeriodicShardSyncManager运行定期同步以查找租赁表中缺少的分片，并根据需要为它们创建租约。

孩子碎片：由LeaseCleanupManager以查看子分片的处理状态，并决定是否可以从租赁表中删除父分片。

ShharDID：分片的 ID。

stream name数据流的标识符采用以下格式：account-id:StreamName:streamCreationTimestamp.

获取正在处理的流的数据的所有分片

从租赁表中获取所有分片租约

筛选出租赁表中没有租约的每个打开的分片

迭代所有找到的打开分片以及每个没有打开父级的打开分片：

在应用程序引导过程中，如果租赁表为空，KCL 会使用ListShardAPI 的过滤选项（ShardFilter可选请求参数），用于仅针对在指定的时间打开的分片的快照检索和创建租赁ShardFilter参数。这些区域有：ShardFilter参数使您能够过滤掉ListShardsAPI。唯一必需的财产ShardFilter参数为Type. KCL 使用Typefilter 属性及其有效值的以下有效值，用于识别并返回可能需要新租用的打开分片的快照：

ShardFilter用于为空租赁表创建租赁，以便为指定的分片快照初始化租赁RetrievalConfig#initialPositionInStreamExtended.

有关 ShardFilter 的更多信息，请参阅 https://docs.amazonaws.cn/kinesis/latest/APIReference/API_ShardFilter.html。

由一名当选的工作人员负责人执行租赁/分片同步以使租赁表与数据流中的最新分片保持最新状态，而不是执行租赁/分片同步的所有工作人员，而是执行租赁/分片同步。

KCL 2.3 使用ChildShards返回的参数GetRecords和SubscribeToShard用于执行租赁/分片同步的 APISHARD_END对于封闭的分片，允许 KCL 工作人员只能为已完成处理的分片的子分片创建租约。对于在整个消费者应用程序中共享，这种租赁/分片同步的优化使用ChildShards的参数GetRecordsAPI。对于专用吞吐量（增强的扇出）消费者应用程序，这种租赁/分片同步的优化使用ChildShards的参数SubscribeToShardAPI。有关更多信息，请参阅 。GetRecords、SubscribeTo分片, 和ChildShard.

随着上述变化，KCL 的行为正从所有员工了解所有现有碎片的模式转变为工人只学习每个员工拥有的碎片的子级碎片的模式。因此，除了在消费者应用程序引导和重新分片事件期间发生的同步之外，KCL 现在还执行额外的定期分片/租赁扫描，以便识别租赁表中的任何潜在漏洞（换句话说，了解所有新分片），以确保完整正在处理数据流的哈希范围，并根据需要为它们创建租赁。PeriodicShardSyncManager是负责运行定期租赁/分片扫描的组件。

有关 的更多信息PeriodicShardSyncManager在 KCL 2.3 中，请参阅https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/软件/亚马逊/kinesis/租赁/LeaseManagementConfig.java #L201-L213.

在 KCL 2.3 中，可以配置新的配置选项PeriodicShardSyncManager在LeaseManagementConfig：

NewCloudWatch现在还可以发出指标来监控PeriodicShardSyncManager. 有关更多信息，请参阅 周期 SHARDSYNC 管理器。

包括优化HierarchicalShardSyncer只为一层碎片创建租约。

在应用程序引导过程中，如果租赁表为空，KCL 会使用ListShardAPI 的过滤选项（ShardFilter可选请求参数），用于仅针对在指定的时间打开的分片的快照检索和创建租赁ShardFilter参数。这些区域有：ShardFilter参数使您能够过滤掉ListShardsAPI。唯一必需的财产ShardFilter参数为Type. KCL 使用Typefilter 属性及其有效值的以下有效值，用于识别并返回可能需要新租用的打开分片的快照：

ShardFilter用于为空租赁表创建租赁，以便为指定的分片快照初始化租赁KinesisClientLibConfiguration#initialPositionInStreamExtended.

有关 ShardFilter 的更多信息，请参阅 https://docs.amazonaws.cn/kinesis/latest/APIReference/API_ShardFilter.html。

由一名当选的工作人员负责人执行租赁/分片同步以使租赁表与数据流中的最新分片保持最新状态，而不是执行租赁/分片同步的所有工作人员，而是执行租赁/分片同步。

KCL 1.14 使用ChildShards返回的参数GetRecords和SubscribeToShard用于执行租赁/分片同步的 APISHARD_END对于封闭的分片，允许 KCL 工作人员只能为已完成处理的分片的子分片创建租约。有关更多信息，请参阅 。GetRecords和ChildShard.

随着上述变化，KCL 的行为正从所有员工了解所有现有碎片的模式转变为工人只学习每个员工拥有的碎片的子级碎片的模式。因此，除了在消费者应用程序引导和重新分片事件期间发生的同步之外，KCL 现在还执行额外的定期分片/租赁扫描，以便识别租赁表中的任何潜在漏洞（换句话说，了解所有新分片），以确保完整正在处理数据流的哈希范围，并根据需要为它们创建租赁。PeriodicShardSyncManager是负责运行定期租赁/分片扫描的组件。

何时KinesisClientLibConfiguration#shardSyncStrategyType设置为ShardSyncStrategyType.SHARD_END、PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold用于确定租赁表中包含漏洞的连续扫描次数的阈值，之后要强制执行分片同步。何时KinesisClientLibConfiguration#shardSyncStrategyType设置为ShardSyncStrategyType.PERIODIC、leasesRecoveryAuditorInconsistencyConfidenceThreshold已忽略。

有关 的更多信息PeriodicShardSyncManager在 KCL 1.14 中，请参阅https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/服务/kinesis/客户库/lib/工作人员/KinesisClientLibConfiguration.java #L987-L999.

在 KCL 1.14 中，新的配置选项可用于配置PeriodicShardSyncManager在LeaseManagementConfig：

NewCloudWatch现在还可以发出指标来监控PeriodicShardSyncManager. 有关更多信息，请参阅 周期 SHARDSYNC 管理器。

KCL 1.14 现在还支持延期租约清理。通过以下方式异步删除租赁LeaseCleanupManager到达后SHARD_END，当分片已超过数据流的保留期之后过期或者因重新分片操作而被关闭时。

新的配置选项如下：LeaseCleanupManager.

包括优化KinesisShardSyncer只为一层碎片创建租约。

MultistreamTracker 界面

要构建可以同时处理多个流的消费者应用程序，必须实现一个名为MultistreamTracker. 此界面包括streamConfigList方法，返回要由 KCL 使用者应用程序处理的数据流及其配置的列表。请注意，正在处理的数据流可以在使用者应用程序运行时期间进行更改。streamConfigListKCL 定期调用，以了解要处理的数据流的变化。

这些区域有：streamConfigList填充StreamConfig列表。

package software.amazon.kinesis.common;

import lombok.Data;
import lombok.experimental.Accessors;

@Data
@Accessors(fluent = true)
public class StreamConfig {
    private final StreamIdentifier streamIdentifier;
    private final InitialPositionInStreamExtended initialPositionInStreamExtended;
    private String consumerArn;
}
                

请注意，StreamIdentifier和InitialPositionInStreamExtended是必填字段，而consumerArn为可选项。您必须提供consumerArn仅当您使用 KCL 2.x 来实施增强型扇出消费者应用程序时才能使用。

有关 的更多信息StreamIdentifier请参阅https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/软件/亚马逊/kinesis/常见/StreamIdentifier.java #L29. 您可以为StreamIdentifier来自序列化的流标识符。序列化的流标识符应采用以下格式：account-id:StreamName:streamCreationTimestamp.

  * @param streamIdentifierSer
     * @return StreamIdentifier
     */
    public static StreamIdentifier multiStreamInstance(String streamIdentifierSer) {
        if (PATTERN.matcher(streamIdentifierSer).matches()) {
            final String[] split = streamIdentifierSer.split(DELIMITER);
            return new StreamIdentifier(split[0], split[1], Long.parseLong(split[2]));
        } else {
            throw new IllegalArgumentException("Unable to deserialize StreamIdentifier from " + streamIdentifierSer);
        }
    }
                

MultistreamTracker还包括在租赁表中删除旧流的租赁的策略（formerStreamsLeasesDeletionStrategy)。请注意，在消费者应用程序运行时期间无法更改策略。有关更多信息，请参阅 。https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/软件/亚马逊/kinesis/处理器/FormerStreamsLeasesDeletionStrategy.java

ConfigsBuilder是一个应用程序范围的类，您可以用它来指定在构建 KCL 消费者应用程序时要使用的所有 KCL 2.x 配置设置。ConfigsBuilder课程现在支持MultistreamTracker接口。你可以初始化ConfigsBuilder要么是使用要从中消耗记录的一个数据流的名称：

 /**
     * Constructor to initialize ConfigsBuilder with StreamName
     * @param streamName
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull String streamName, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.right(streamName);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }

或者你可以初始化ConfigsBuilder和MultiStreamTracker如果你想实现一个同时处理多个流的 KCL 消费者应用程序。

* Constructor to initialize ConfigsBuilder with MultiStreamTracker
     * @param multiStreamTracker
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull MultiStreamTracker multiStreamTracker, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.left(multiStreamTracker);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }                    
                

随着为 KCL 消费者应用程序实施了多流支持，应用程序的租赁表中的每一行现在都包含该应用程序处理的多个数据流的分片 ID 和流名称。

当实施对 KCL 消费者应用程序的多流支持时，leaseKey 将采用以下结构：account-id:StreamName:streamCreationTimestamp:ShardId. 例如，111111111:multiStreamTest-1:12345:shardId-000000000336.