KCL1.x 和 2.x 信息

KCL1.x

有关更多信息，请参阅 培养 KCL 1.x 消费者

KCL2.x

有关更多信息，请参阅 培养 KCL 2.x 消费者

KCL消费者应用程序 — 一种使用KCL数据流定制的应用程序，旨在读取和处理数据流中的记录。

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

Worker — 使用KCL者应用程序实例用来开始处理数据的高级类。

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

租约 – 定义工作程序和分片之间绑定的数据。分布式KCL消费者应用程序使用租约将数据记录处理分成一群工作人员。在任何给定时间，每个数据记录分片都通过leaseKey变量标识的租约绑定到特定的工作人员。

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

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

有关更多信息，（这些是 Java KCL 存储库），请参阅 KCL 1.x 版的 https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java 和适用于 2.x 的 https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java。 KCL

租赁表-一个唯一的 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使用ListShardAPI的筛选选项（ShardFilter可选的请求参数）来检索和创建租约，仅针对在参数指定时间打开的分片的快照进行检索和创建租约。ShardFilter该ShardFilter参数使您可以过滤掉的响应ListShardsAPI。该ShardFilter参数的唯一必需属性是Type。 KCL使用 Type filter 属性及其以下有效值来识别并返回可能需要新租约的已打开分片的快照：

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

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

而不是所有工作人员都在执行同lease/shard synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard步。

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

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

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

在 KCL 2.3 中，新的配置选项可在以下版本PeriodicShardSyncManager中进行配置LeaseManagementConfig：

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

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

在应用程序的引导过程中，如果租赁表为空，则KCL使用ListShardAPI的筛选选项（ShardFilter可选的请求参数）来检索和创建租约，仅针对在参数指定时间打开的分片的快照进行检索和创建租约。ShardFilter该ShardFilter参数使您可以过滤掉的响应ListShardsAPI。该ShardFilter参数的唯一必需属性是Type。 KCL使用 Type filter 属性及其以下有效值来识别并返回可能需要新租约的已打开分片的快照：

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

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

而不是所有工作人员都在执行同lease/shard synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard步。

KCL1.14 使用GetRecords和的ChildShards返回参数对已关闭的SubscribeToShardAPIs分片执行租用/分片同步，从而允许KCL工作程序仅SHARD_END为其完成处理的分片的子分片创建租约。有关更多信息，请参阅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/KinesisClientLibConfiguration.java #L987-L999。

在 KCL 1.14 中，新的配置选项可在以下版本PeriodicShardSyncManager中LeaseManagementConfig进行配置：

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

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

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

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

MultistreamTracker 接口

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

该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/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java #L129. 要创建 StreamIdentifier，我们建议您从 v2.5.0 及更高版本提供的 streamArn 和 streamCreationEpoch 创建一个多流实例。在不支持的 v2. KCL 3 和 v2.4 中streamArm，使用格式创建多流实例。account-id:StreamName:streamCreationTimestamp从下一个主要版本开始，此格式将弃用且不再受支持。

MultistreamTracker 还包括可删除租约表中旧流租约的策略（formerStreamsLeasesDeletionStrategy）。请注意，策略CANNOT将在使用者应用程序运行时进行更改。欲了解更多信息，请参阅 https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b//amazon-kinesis-client.java src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy

ConfigsBuilder是一个应用程序范围的类，可用于指定构建使用者应用程序时要使用的所有 KCL 2.x 配置设置。KCL 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如果你想实现一个同时处理多个流的 ConfigsBuilder 使用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。