實施生產者

private static void sendStockTrade(StockTrade trade, KinesisAsyncClient kinesisClient,
            String streamName) {
        byte[] bytes = trade.toJsonAsBytes();
        // The bytes could be null if there is an issue with the JSON serialization by the Jackson JSON library.
        if (bytes == null) {
            LOG.warn("Could not get JSON bytes for stock trade");
            return;
        }

        LOG.info("Putting trade: " + trade.toString());
        PutRecordRequest request = PutRecordRequest.builder()
                .partitionKey(trade.getTickerSymbol()) // We use the ticker symbol as the partition key, explained in the Supplemental Information section below.
                .streamName(streamName)
                .data(SdkBytes.fromByteArray(bytes))
                .build();
        try {
            kinesisClient.putRecord(request).get();
        } catch (InterruptedException e) {
            LOG.info("Interrupted, assuming shutdown.");
        } catch (ExecutionException e) {
            LOG.error("Exception while sending data to Kinesis. Will try again next cycle.", e);
        }
    }
                    

請參閱以下的程式碼詳解：

• PutRecordAPI期望一個字節數組，並且您必須將交易轉換為JSON格式。這一行程式碼將執行該項操作：

  
  byte[] bytes = trade.toJsonAsBytes();
                              

• 傳送交易之前，您必須先建立新的 PutRecordRequest 執行個體 (本例中稱為要求)。每個 request 都需要串流名稱、分割區索引鍵和資料 Blob。

  
  PutPutRecordRequest request = PutRecordRequest.builder()
      .partitionKey(trade.getTickerSymbol()) // We use the ticker symbol as the partition key, explained in the Supplemental Information section below.
      .streamName(streamName)
      .data(SdkBytes.fromByteArray(bytes))
      .build();
                              

  此範例使用股票股票代碼做為分割索引鍵，該索引鍵會將記錄對應至特定的碎片。實際上，每個碎片應該會有成千上百的分割區索引鍵，使記錄均勻地分佈於串流中。如需如何加入資料至串流的詳細資訊，請參閱將資料寫入 Amazon Kinesis Data Streams。

  現在 request 已準備好傳送至用戶端 (put 操作)：

  
     kinesisClient.putRecord(request).get();
                              

• 錯誤檢查和日誌記錄肯定是頗為實用的附加功能。此程式碼將記錄錯誤情況：

  
  if (bytes == null) {
      LOG.warn("Could not get JSON bytes for stock trade");
      return;
  }
  

  在 put 操作的周圍加入 try/catch 區塊：

  
  try {
   	kinesisClient.putRecord(request).get();
  } catch (InterruptedException e) {
              LOG.info("Interrupted, assuming shutdown.");
  } catch (ExecutionException e) {
              LOG.error("Exception while sending data to Kinesis. Will try again next cycle.", e);
  }
                              

  這麼做是因為 Kinesis Data Streams put 操作可能由於網路錯誤或是資料串流達到其傳輸量限制並受到調節而導致失敗。建議您仔細考慮put作業的重試原則，以避免資料遺失，例如使用重試。

• 狀態記錄也很實用，但可有可無：

  
  LOG.info("Putting trade: " + trade.toString());