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

# 實作消費者
<a name="tutorial-stock-data-kplkcl2-consumer"></a>

本教學課程中的消費者應用程式會持續處理您的資料串流中的股票交易。隨後，其將輸出每分鐘買進和賣出最多的熱門股票。此應用程式是使用 Kinesis Client Library (KCL) 所建置，由該程式庫執行取用者應用程式常見的諸多繁重工作。如需詳細資訊，請參閱[KCL 1.x 和 2.x 資訊](shared-throughput-kcl-consumers.md)。

請查看原始碼並對照檢閱以下資訊。

**StockTradesProcessor 類別**  
提供給您的消費者主要類別，其會執行下列任務：  
+ 讀取做為引數傳入的應用程式、資料串流和區域名稱。
+ 使用區域名稱建立`KinesisAsyncClient`執行個體。
+ 建立 `StockTradeRecordProcessorFactory` 執行個體以提供由 `ShardRecordProcessor` 執行個體實作的 `StockTradeRecordProcessor` 執行個體。
+ 使用 `KinesisAsyncClient`、`ApplicationName`、 `StreamName`和 `ConfigsBuilder`執行個體建立`StockTradeRecordProcessorFactory`執行個體。這對於使用預設值建立所有組態非常有用。
+ 使用 `ConfigsBuilder` 執行個體建立 KCL 排程器 (之前，在 KCL 版本 1.x 中，它被稱為 KCL 工作者)。
+ 排程器會為每個碎片 (已指派給此消費者執行個體) 建立新的執行緒，以持續循環從資料串流讀取記錄。接著，其將叫用 `StockTradeRecordProcessor` 執行個體以處理收到的各個批次記錄。

**StockTradeRecordProcessor 類別**  
`StockTradeRecordProcessor` 執行個體的實作，而此執行個體將實作五個必要的方法：`initialize`、`processRecords`、`leaseLost`、`shardEnded` 和 `shutdownRequested`。  
KCL 使用 `initialize` 和 `shutdownRequested` 方法，讓記錄處理器得知何時應準備好開始接收記錄以及何時應停止接收記錄，好讓程式庫能夠執行任何應用程式特定的設定和終止任務。`leaseLost` 和 `shardEnded` 則用於實作當遺失租約或處理已達碎片結尾時應執行什麼操作的任何邏輯。在此範例中，我們只記錄指出這些事件的訊息。  
我們會提供這些方法的程式碼。主要處理任務在 `processRecords` 方法中進行，而此方法將使用 `processRecord` 處理每筆記錄。後一種方法以幾乎全空的架構程式碼提供，讓您於下一個步驟進行實作，屆時將會有更詳細的說明。  
另請注意 `processRecord` 支援方法的實作：`reportStats` 和 `resetStats`，其最初的原始碼為全空。  
程式碼已為您實作 `processRecords` 方法，並將執行以下步驟：  
+ 對每一筆傳入的記錄，它會呼叫其上的 `processRecord`。
+ 若自從上次報告後已歷時至少 1 分鐘，請先呼叫 `reportStats()` 列印出最新統計資料，接著呼叫 `resetStats()` 清除統計資料以使下一個間隔僅包含新記錄。
+ 設定下一次報告時間。
+ 若自從最後一個檢查點過後已歷時至少 1 分鐘，請呼叫 `checkpoint()`。
+ 設定下一次檢查點作業時間。
此方法使用 60 秒的間隔做為報告及檢查點作業率。如需有關檢查點的詳細資訊，請參閱[使用 Kinesis Client Library](https://docs.aws.amazon.com/streams/latest/dev/shared-throughput-kcl-consumers.html)。

**StockStats 類別**  
此類別針對一段時間內最熱門的股票提供資料保留與統計資料追蹤。其程式碼已為您提供且包含下列方法：  
+ `addStockTrade(StockTrade)`：將給定的 `StockTrade` 注入目前統計資料。
+ `toString()`：以格式化字串的形式傳回統計資料。
此類別透過保留每個股票交易總數的執行中計數和最大計數，來追蹤最熱門的股票。每當股票交易達成時，其將更新這些計數。

為 `StockTradeRecordProcessor` 類別的各個方法加入程式碼，如以下步驟所示。

**實作消費者**

1. 實作 `processRecord` 方法，藉此執行個體化正確大小的 `StockTrade` 物件並將記錄資料加入該物件，且於發生問題時記錄警告。

   ```
   byte[] arr = new byte[record.data().remaining()];
   record.data().get(arr);
   StockTrade trade = StockTrade.fromJsonAsBytes(arr);
       if (trade == null) {
           log.warn("Skipping record. Unable to parse record into StockTrade. Partition Key: " + record.partitionKey());
           return;
           }
   stockStats.addStockTrade(trade);
   ```

1. 實作 `reportStats`方法。修改輸出格式以符合您的偏好設定。

   ```
   System.out.println("****** Shard " + kinesisShardId + " stats for last 1 minute ******\n" +
   stockStats + "\n" +
   "****************************************************************\n");
   ```

1. 實作 `resetStats` 方法以便建立新的 `stockStats` 執行個體。

   ```
   stockStats = new StockStats();
   ```

1. 實作`ShardRecordProcessor`界面所需的下列方法：

   ```
   @Override
   public void leaseLost(LeaseLostInput leaseLostInput) {
       log.info("Lost lease, so terminating.");
   }
   
   @Override
   public void shardEnded(ShardEndedInput shardEndedInput) {
       try {
           log.info("Reached shard end checkpointing.");
           shardEndedInput.checkpointer().checkpoint();
       } catch (ShutdownException | InvalidStateException e) {
           log.error("Exception while checkpointing at shard end. Giving up.", e);
       }
   }
   
   @Override
   public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
       log.info("Scheduler is shutting down, checkpointing.");
       checkpoint(shutdownRequestedInput.checkpointer());
   }
   
   private void checkpoint(RecordProcessorCheckpointer checkpointer) {
       log.info("Checkpointing shard " + kinesisShardId);
       try {
           checkpointer.checkpoint();
       } catch (ShutdownException se) {
           // Ignore checkpoint if the processor instance has been shutdown (fail over).
           log.info("Caught shutdown exception, skipping checkpoint.", se);
       } catch (ThrottlingException e) {
           // Skip checkpoint when throttled. In practice, consider a backoff and retry policy.
           log.error("Caught throttling exception, skipping checkpoint.", e);
       } catch (InvalidStateException e) {
           // This indicates an issue with the DynamoDB table (check for table, provisioned IOPS).
           log.error("Cannot save checkpoint to the DynamoDB table used by the Amazon Kinesis Client Library.", e);
       }
   }
   ```

**執行消費者**

1. 執行您在[[實作生產者](tutorial-stock-data-kplkcl2-producer.md)實作生產者](tutorial-stock-data-kplkcl2-producer.md)時撰寫的生產者，將模擬的股票交易記錄注入您的串流。

1. 確認稍早 (建立 IAM 使用者 使用者時) 擷取的存取金鑰和私密金鑰對是否已儲存至 `~/.aws/credentials` 檔案。

1. 使用以下引數執行 `StockTradesProcessor` 類別：

   ```
   StockTradesProcessor StockTradeStream us-west-2
   ```

   請注意，如果您是在 `us-west-2` 以外的區域建立串流，則此處必須改為指定該區域。

一分鐘後，您應會看到類似以下內容的輸出，而且此後每分鐘將重新整理一次輸出：

```
  
  ****** Shard shardId-000000000001 stats for last 1 minute ******
  Most popular stock being bought: WMT, 27 buys.
  Most popular stock being sold: PTR, 14 sells.
  ****************************************************************
```

## 後續步驟
<a name="tutorial-stock-data-kplkcl2-consumer-next"></a>

[（選用） 擴展消費者](tutorial-stock-data-kplkcl2-consumer-extension.md)