使用 Kinesis Client Library

连接到数据流

枚举数据流中的分片

使用租约来协调分片与其工作程序的关联

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

从数据流中提取数据记录

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

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

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

KCL 1.x

有关更多信息，请参阅 开发 KCL 1.x 消费端。

KCL 2.x

有关更多信息，请参阅 开发 KCL 2.x 消费端。

KCL 消费端应用程序 – 使用 KCL 自定义构建的应用程序，旨在读取和处理数据流中的记录。

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

工作程序 – KCL 消费端应用程序实例用来开始处理数据的高级类。

工作程序负责初始化和监督各种任务，包括同步分片和租约信息、跟踪分片分配以及处理来自分片的数据。工作程序向 KCL 提供消费端应用程序的配置信息，例如此 KCL 消费端应用程序要处理的数据记录的数据流名称以及访问此数据流所需的 Amazon 凭证。工作程序还会启动特定的 KCL 消费端应用程序实例，将数据记录从数据流传输到记录处理器。

租约 – 定义工作程序和分片之间绑定的数据。分布式 KCL 消费端应用程序使用租约在一组工作程序中对数据记录进行分区。在任何指定时间，每个数据记录分片都通过由 leaseKey 变量标识的租约绑定到特定的工作程序。

默认情况下，工作人员可以同时持有一份或多份租约（以 maxLeasesForWorker 变量的值为准）。

例如，如果您的消费端应用程序实例 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/amazonaws/services/kinesis/Leases/impl/lease.java for KCL 1.x 和 KCL 2.x 的 https://github.com/awslabs/ /blob/ma ster/ /src/main/java/software/amazon/kinesis/leases/Lease.java amazon-kinesis-client amazon-kinesis-client

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

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

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

checkpointSubSequence数字：使用 Kinesis Producer 库的聚合功能时，这是对检查点的扩展，用于跟踪 Kinesis 记录中的单个用户记录。

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

leaseKey：租赁的唯一标识符。每份租约都是数据流中一个分片所特有的，一份由一个工作程序持有。

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

ownerSwitchesSince检查点：自上次写检查点以来，这份租约更换了多少次员工。

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

hashrange：PeriodicShardSyncManager 用来运行定期同步，以查找租约表中缺少的分片，并在需要时为分片创建租约。

childshards：LeaseCleanupManager 用来查看子分片的处理状态并决定是否可以从租约表中删除父分片。

shardID：分片的 ID。

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

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

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

筛选出租约表中没有租约的所有开放分片

迭代所有找到的开放分片以及没有开放父分片的所有开放分片：

在应用程序的引导启动过程中，如果租约表为空，KCL 将利用 ListShard API 的筛选选项（ShardFilter 可选请求参数），仅针对在 ShardFilter 参数指定时间开放的分片的快照检索和创建租约。ShardFilter 参数可以让您筛选出 ListShards API 的响应。ShardFilter 参数唯一需要的属性是 Type。KCL 使用 Type 筛选属性及其以下有效值来识别并返回可能需要新租约的开放分片的快照：

ShardFilter 用于为空租约表创建租约，以初始化在 RetrievalConfig#initialPositionInStreamExtended 中指定的分片快照的租约。

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

要让租约表与数据流中的最新分片保持同步，不是所有工作程序都执行租约/分片同步，而是由一个所选的主工作程序来执行租约/分片同步。

KCL 2.3 使用 GetRecords 和 SubscribeToShard API 的 ChildShards 返回参数在 SHARD_END 时对关闭的分片执行租约/分片同步，从而允许 KCL 工作程序仅为其完成处理的分片的子分片创建租约。对于共享吞吐量消费端应用程序，租约/分片同步的这种优化使用了 GetRecords API 的 ChildShards 参数。对于专用吞吐量（增强型扇出）消费端应用程序，租约/分片同步的这种优化使用 SubscribeToShard API 的 ChildShards 参数。有关更多信息，请参阅GetRecordsSubscribeToShards、和ChildShard。

经过上述更改，KCL 的行为正在从所有工作程序学习所有现有分片的模式，转变为工作程序只学习每个工作程序拥有的分片的子分片的模式。因此，除了在消费端应用程序引导启动和重新分片事件期间发生的同步外，KCL 现在还会执行额外的定期分片/租约扫描，从而识别租约表中的任何潜在漏洞（也就是了解所有新分片），确保数据流的完整哈希范围得到处理，并在需要时为它们创建租约。PeriodicShardSyncManager 是负责定期运行租约/分片扫描的组件。

有关 KCL 2.3 PeriodicShardSyncManager 中的更多信息，请参阅 https://github.com/awslabs/ amazon-kinesis-client /blob/master/ /src/main/java/sroftware/amazon/kinesis/leases/.jav amazon-kinesis-client a #L201-L213。LeaseManagementConfig

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

现在还会发布新的 CloudWatch 指标来监控的PeriodicShardSyncManager运行状况。有关更多信息，请参阅 PeriodicShardSyncManager。

包括对 HierarchicalShardSyncer 的优化，可仅为一层分片创建租约。

在应用程序的引导启动过程中，如果租约表为空，KCL 将利用 ListShard API 的筛选选项（ShardFilter 可选请求参数），仅针对在 ShardFilter 参数指定时间开放的分片的快照检索和创建租约。ShardFilter 参数可以让您筛选出 ListShards API 的响应。ShardFilter 参数唯一需要的属性是 Type。KCL 使用 Type 筛选属性及其以下有效值来识别并返回可能需要新租约的开放分片的快照：

ShardFilter 用于为空租约表创建租约，以初始化在 KinesisClientLibConfiguration#initialPositionInStreamExtended 中指定的分片快照的租约。

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

要让租约表与数据流中的最新分片保持同步，不是所有工作程序都执行租约/分片同步，而是由一个所选的主工作程序来执行租约/分片同步。

KCL 1.14 使用 GetRecords 和 SubscribeToShard API 的 ChildShards 返回参数在 SHARD_END 时对关闭的分片执行租约/分片同步，从而允许 KCL 工作程序仅为其完成处理的分片的子分片创建租约。有关更多信息，请参阅 GetRecords和 ChildShard。

经过上述更改，KCL 的行为正在从所有工作程序学习所有现有分片的模式，转变为工作程序只学习每个工作程序拥有的分片的子分片的模式。因此，除了在消费端应用程序引导启动和重新分片事件期间发生的同步外，KCL 现在还会执行额外的定期分片/租约扫描，从而识别租约表中的任何潜在漏洞（也就是了解所有新分片），确保数据流的完整哈希范围得到处理，并在需要时为它们创建租约。PeriodicShardSyncManager 是负责定期运行租约/分片扫描的组件。

如果 KinesisClientLibConfiguration#shardSyncStrategyType 设置为 ShardSyncStrategyType.SHARD_END，则 PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold 用于确定租约表中包含漏洞的连续扫描次数的阈值，之后将强制执行分片同步。当 KinesisClientLibConfiguration#shardSyncStrategyType 设置为时 ShardSyncStrategyType.PERIODIC，leasesRecoveryAuditorInconsistencyConfidenceThreshold 将被忽略。

有关 KCL 1.14 PeriodicShardSyncManager 中的更多信息，请参阅 https://github.com/awslabs/ amazon-kinesis-client /blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ .java #L987-L999 KinesisClientLibConfiguration。

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

现在还会发布新的 CloudWatch 指标来监控的PeriodicShardSyncManager运行状况。有关更多信息，请参阅 PeriodicShardSyncManager。

KCL 1.14 现在还支持延期租约清理。当分片超过数据流的保留期或因重新分片操作而关闭时，LeaseCleanupManager 会在到达 SHARD_END 时异步删除租约。

可以使用新的配置选项来配置 LeaseCleanupManager。

包括对 KinesisShardSyncer 的优化，可仅为一层分片创建租约。

MultistreamTracker 接口

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

该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 是选填字段。只有在使用 KCL 2.x 实现增强型扇出消费端应用程序时，才必须提供 consumerArn。

有关更多信息，请参阅 https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/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/ /src/main/java/sortware/am amazon-kinesis-client azon/kinesis/processor/.java FormerStreamsLeasesDeletionStrategy

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;
    }

或者，MultiStreamTracker如果要实现同时处理多个流的 KCL 使用者应用程序，也可以使用进行初始化 ConfigsBuilder 。

* 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。