本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

# 實作消費者取消彙總
<a name="kinesis-kpl-consumer-deaggregation"></a>

自發行版本 1.4.0 起，KCL 支援自動取消彙整 KPL 使用者記錄。在您更新 KCL 後，使用舊版 KCL 撰寫的取用者應用程式將編譯程式碼而不會進行任何修改。不過，如果生產者端使用了 KPL 彙整，則有一項與檢查點作業相關的細微之處須留意：彙整的記錄中所有的子記錄具有相同的序號，所以若您需要區別子記錄，就必須隨檢查點存放額外的資料。這些額外的資料稱為*子序號*。

**Topics**
+ [從舊版 KCL 遷移](#kinesis-kpl-consumer-deaggregation-migration)
+ [使用 KPL 去彙總的 KCL 擴充功能](#kinesis-kpl-consumer-deaggregation-extensions)
+ [直接使用 GetRecords](#kinesis-kpl-consumer-deaggregation-getrecords)

## 從舊版 KCL 遷移
<a name="kinesis-kpl-consumer-deaggregation-migration"></a>

您不需要變更現有的呼叫，即可使用彙總執行檢查點。保證您仍能成功擷取存放於 Kinesis Data Streams 的所有記錄。KCL 現在提供兩個新的檢查點操作，以支援特定使用案例，如下所述。

如果您在 KPL 支援之前為 KCL 編寫現有程式碼，而且您的檢查點操作在沒有引數的情況下呼叫，則相當於對批次中最後一個 KPL 使用者記錄的序號進行檢查點。如果使用序號字串呼叫檢查點操作，則等同於對批次的指定序號及隱含的子序號 0 (零) 執行檢查點作業。

呼叫新的 KCL 檢查點操作而不帶任何引數如 `checkpoint()` 語意上等同於對批次中其上次 `Record` 呼叫的序號及隱含的子序號 0 (零) 執行檢查點作業。

呼叫新的 KCL 檢查點操作如 `checkpoint(Record record)` 語意上等同於對指定 `Record` 的序號及隱含的子序號 0 (零) 執行檢查點作業。若 `Record` 呼叫實際為 `UserRecord`，則會對 `UserRecord` 序號和子序號執行檢查點作業。

呼叫新的 KCL 檢查點操作如 `checkpoint(String sequenceNumber, long subSequenceNumber)` 會對指定的序號及指定的子序號執行明示檢查點作業。

上述任何情況下，當檢查點已存放於 Amazon DynamoDB 檢查點資料表之後，KCL 便能正確地恢復擷取記錄，就算應用程式當機並重新啟動也沒問題。如果序列中包含多筆記錄，則會從序號最近執行過檢查點作業的記錄中為下一個子序號的記錄開始擷取。若最近的檢查點包括前一序號記錄的最新子序號，將從下一個序號的記錄開始擷取。

下一節討論了消費者的序列和子序列檢查點的詳細資訊，必須避免略過和重複記錄。若停止並重新啟動消費者的記錄處理會略過 (或重複) 記錄無關緊要，您就可以執行現有的程式碼而無須修改。

## 使用 KPL 去彙總的 KCL 擴充功能
<a name="kinesis-kpl-consumer-deaggregation-extensions"></a>

KPL 去彙總可能涉及子序列檢查點。為方便使用子序列檢查點作業，KCL 增加了 `UserRecord` 類別：

```
public class UserRecord extends Record {     
    public long getSubSequenceNumber() {
    /* ... */
    }      
    @Override 
    public int hashCode() {
    /* contract-satisfying implementation */ 
    }      
    @Override 
    public boolean equals(Object obj) {
    /* contract-satisfying implementation */ 
    } 
}
```

現已使用此類別代替 `Record`。這不會破壞現有的程式碼，因為其為 `Record` 的子類別。`UserRecord` 類別同時代表實際的子記錄和未彙整的標準記錄。未彙整的記錄可想像成恰有一筆子記錄的已彙整記錄。

此外，`IRecordProcessorCheckpointer` 也增加了兩項新的操作：

```
public void checkpoint(Record record); 
public void checkpoint(String sequenceNumber, long subSequenceNumber);
```

若要開始使用子序號檢查點作業，您可進行以下轉換。更改以下形式的程式碼：

```
checkpointer.checkpoint(record.getSequenceNumber());
```

新形式的程式碼：

```
checkpointer.checkpoint(record);
```

建議您使用 `checkpoint(Record record)` 形式執行子序列檢查點作業。不過，若您已將 `sequenceNumbers` 存放在字串中用於檢查點作業，則現在亦應存放 `subSequenceNumber`，如以下範例所示：

```
String sequenceNumber = record.getSequenceNumber(); 
long subSequenceNumber = ((UserRecord) record).getSubSequenceNumber();  // ... do other processing  
checkpointer.checkpoint(sequenceNumber, subSequenceNumber);
```

從 轉換為`UserRecord` `Record`一律會成功，因為實作一律使用 `UserRecord`。除非需要對序號進行算術運算，否則這種方式並不建議。

處理 KPL 使用者記錄時，KCL 會將子序號寫入 Amazon DynamoDB 成為每一列的額外欄位。舊版 KCL 是使用 `AFTER_SEQUENCE_NUMBER` 在恢復檢查點作業時擷取記錄。目前有 KPL 支援的 KCL 則改用 `AT_SEQUENCE_NUMBER`。在擷取已對序號執行過檢查點作業的記錄時，會檢查已執行檢查點作業的子序號，且將視需要刪除子記錄 (若已對最後一筆子記錄執行檢查點作業，則可能全部刪除)。同樣地，未彙整的記錄可想像成只有一筆子記錄的已彙整記錄，所以同一套演算法對已彙整和未彙整的記錄都適用。

## 直接使用 GetRecords
<a name="kinesis-kpl-consumer-deaggregation-getrecords"></a>

您也可以選擇不使用 KCL 而是直接調用 API 操作 `GetRecords` 來擷取 Kinesis Data Streams 記錄。若要將這些擷取到的記錄解壓縮為原始 KPL 使用者記錄，請呼叫 `UserRecord.java` 的以下任一項靜態操作：

```
public static List<Record> deaggregate(List<Record> records)

public static List<UserRecord> deaggregate(List<UserRecord> records, BigInteger startingHashKey, BigInteger endingHashKey)
```

第一項操作使用 `0` 的預設值 `startingHashKey` (零) 以及 `2^128 -1` 的預設值 `endingHashKey`。

上述每一項操作都將取消彙總指定的 Kinesis Data Streams 記錄清單，轉成 KPL 使用者記錄清單。傳回的記錄清單將會捨棄顯式雜湊索引鍵或分割區索引鍵落在 `startingHashKey` (含) 以及 `endingHashKey` (含) 範圍外的任何 KPL 使用者記錄。