KCL 1.x 和 2.x 資訊

KCL 1.x

如需詳細資訊，請參閱 開發 KCL 1.x 消費者

KCL 2.x

如需詳細資訊，請參閱 開發 KCL 2.x 消費者

KCL 取用者應用程式 – 使用 自訂建置的應用程式，KCL旨在從資料串流讀取和處理記錄。

取用者應用程式執行個體 - 取用KCL者應用程式通常會分散，其中一或多個應用程式執行個體會同時執行，以便協調故障和動態載入平衡資料記錄處理。

工作者 – KCL取用者應用程式執行個體用來開始處理資料的高階類別。

工作者會初始化並監督各種任務，包括同步處理碎片和租用資訊、追蹤碎片指派，以及處理來自碎片的資料。工作者KCL會提供取用者應用程式的組態資訊，例如資料串流的名稱，其資料記錄此KCL取用者應用程式將要處理的資料，以及存取此資料串流所需的 AWS 登入資料。工作者也會啟動該特定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/amazonaws/services/kinesis/leases/impl/Lease.java for KCL 1.x 和 https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java for KCL 2.x。

租用資料表 - 唯一的 Amazon DynamoDB 資料表，用於追蹤資料串流中的碎片，該KDS資料串流由KCL消費者應用程式的工作者租用和處理。租用資料表必須與KCL消費者應用程式執行時來自資料串流的最新碎片資訊保持同步 （在工作者內和所有工作者之間）。如需詳細資訊，請參閱使用租用資料表追蹤KCL消費者應用程式處理的碎片。

記錄處理器 – 定義KCL消費者應用程式如何處理從資料串流取得之資料的邏輯。在執行時間，KCL消費者應用程式執行個體會執行個體化工作者，而此工作者會執行個體化其持有租用的每個碎片的一個記錄處理器。

checkpoint：碎片的最新檢查點序號。資料串流中所有碎片的此值皆為獨一無二。

checkpointSubSequence號碼：使用 Kinesis Producer Library 的彙總功能時，這是檢查點的延伸，可追蹤 Kinesis 記錄中的個別使用者記錄。

leaseCounter：用於租用版本控制，以便工作者可以偵測他們的租用已由其他工作者所取得。

leaseKey：租用的唯一識別符。每項租用特屬於資料串流中的某個碎片，一次由一個工作者所持有。

leaseOwner：保留此租用的工作者。

ownerSwitchesSince檢查點：自上次撰寫檢查點以來，此租用變更工作者的次數。

parentShardId：用於確保在子碎片上開始處理之前，父碎片已完全處理。這可確保按照記錄放入串流中的相同順序處理記錄。

hashrange：PeriodicShardSyncManager 用於執行週期性同步以尋找租用資料表中遺失的碎片，並在需要時為其建立租用。

childshards：LeaseCleanupManager 用於檢閱子碎片的處理狀態，並決定是否可以從租用資料表中刪除父碎片。

shardID：碎片的 ID。

串流名稱資料串流的識別碼，格式如下：account-id:StreamName:streamCreationTimestamp。

擷取正在處理的資料串流的所有碎片

從租用資料表中擷取所有碎片租用

篩選出租用資料表中沒有租用的每個開放碎片

逐一查看所有找到的開放碎片以及每個沒有開放父級的開放碎片：

在應用程式的引導期間，如果租用資料表是空的， KCL會使用 ListShard API的篩選選項 (ShardFilter選用的請求參數） 來擷取和建立租用，僅限在 ShardFilter 參數指定的時間開啟的碎片快照。ShardFilter 參數可讓您篩選掉 ListShards 的回應API。ShardFilter 參數的唯一必要屬性是 Type。 KCL使用Type篩選條件屬性及其下列有效值來識別和傳回可能需要新租用的開啟碎片快照：

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同步的工作者。

KCL 2.3 使用 GetRecords和 的ChildShards傳回參數SubscribeToShardAPIs，對封閉碎片執行SHARD_END在 發生的租用/碎片同步，允許KCL工作者僅為已完成處理之碎片的子碎片建立租用。對於在整個取用者應用程式中共用，此租用/碎片同步的最佳化會使用 GetRecords 的 ChildShards 參數API。對於專用輸送量 （增強的扇出） 消費者應用程式，此租用/碎片同步最佳化會使用 SubscribeToShard 的 ChildShards 參數API。如需詳細資訊，請參閱GetRecords、SubscribeToShards及ChildShard。

隨著上述變更， 的行為KCL會從學習所有現有碎片的所有員工模型，轉移到僅學習每個工作者所擁有碎片子碎片的工作者模型。因此，除了在消費者應用程式引導和重新碎片事件期間發生的同步之外， KCL 現在還會執行額外的定期碎片/碎片掃描，以識別租用資料表中的任何潛在孔洞 （換言之，了解所有新碎片），以確保資料處理串流的完整雜湊範圍，並視需要為其建立租用。 PeriodicShardSyncManager 是負責執行定期租用/碎片掃描的元件。

如需 2.3 KCL 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會使用 ListShard API的篩選選項 (ShardFilter選用的請求參數） 來擷取和建立租用，僅限在 ShardFilter 參數指定的時間開啟的碎片快照。ShardFilter 參數可讓您篩選掉 ListShards 的回應API。ShardFilter 參數的唯一必要屬性是 Type。 KCL使用Type篩選條件屬性及其下列有效值來識別和傳回可能需要新租用的開啟碎片快照：

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同步的工作者。

KCL 1.14 使用 GetRecords和 的ChildShards傳回參數SubscribeToShardAPIs，對封閉碎片執行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/KinesisClientLibConfiguration.java#L987-L999。

在 KCL 1.14 版中，新的組態選項可用於在 PeriodicShardSyncManager中設定 LeaseManagementConfig：

現在也會發出新的 CloudWatch 指標來監控 的運作狀態PeriodicShardSyncManager。如需詳細資訊，請參閱PeriodicShardSyncManager。

KCL 1.14 現在也支援延遲租用清理。當碎片超過資料串流的保留期限或因重新分片操作而關閉時，達到 SHARD_END 時，LeaseCleanupManager 會以非同步方式刪除租用。

新組態選項可用於設定 LeaseCleanupManager：

包括 KinesisShardSyncer 的最佳化，以僅為一層碎片建立租用。

MultistreamTracker 介面

若要建置可同時處理多個串流的消費者應用程式，您必須實作稱為 的新界面MultistreamTracker。此界面包含傳回資料串流及其組態清單streamConfigList的方法，以供KCL取用者應用程式處理。請注意，正在處理的資料串流可以在取用者應用程式執行階段期間變更。 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，我們建議您從 streamArn和 streamCreationEpoch v2.5.0 及更新版本中提供的 建立多串流執行個體。在不支援 的 v2KCL.3 和 v2.4 中streamArm，使用 格式 建立多串流執行個體。 account-id:StreamName:streamCreationTimestamp從下一個主要版本開始，此格式將被取代，不再受支援。

MultistreamTracker 還包括刪除租用資料表 (formerStreamsLeasesDeletionStrategy) 中舊串流的租用的策略。請注意，策略CANNOT會在取用者應用程式執行時間期間變更。如需詳細資訊，請參閱 https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/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。