プロデューサーを実装する

次のコードの詳細を参照してください。

• PutRecord API はバイト配列を想定するため、trade を JSON 形式に変換する必要があります。この操作は、次の 1 行のコードによって行われます。

  byte[] bytes = trade.toJsonAsBytes();

• 取引を送信する前に、新しい PutRecordRequest インスタンス (この場合、putRecord と呼ばれる) を作成する必要があります。

  PutRecordRequest putRecord = new PutRecordRequest();

  各 PutRecord の呼び出しには、ストリーム名、パーティションキー、およびデータ BLOB が必要です。次のコードによって、putRecord メソッドを使用して、これらのフィールドを setXxxx() オブジェクトに追加します。

  putRecord.setStreamName(streamName);
  putRecord.setPartitionKey(trade.getTickerSymbol());
  putRecord.setData(ByteBuffer.wrap(bytes));

  この例では、株式チケットをパーティションキーとして使用することで、レコードを特定のシャードにマッピングしています。実際には、レコードがストリーム全体に均等に分散するように、シャード 1 つあたりに数百個または数千個のパーティションキーを用意する必要があります。ストリームにデータを追加する方法の詳細については、ストリームにデータを追加するを参照してください。

  次に、putRecord をクライアントに送信 (put オペレーション) することができます。

  kinesisClient.putRecord(putRecord);

• エラーチェックとログ記録は、いつでも追加して損はありません。次のコードによって、エラー状態を記録します。

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

  put オペレーションの前後に try/catch ブロックを追加します。

  try {
         kinesisClient.putRecord(putRecord);
  } catch (AmazonClientException ex) {
         LOG.warn("Error sending record to Amazon Kinesis.", ex);
  }

  これは、ネットワークエラーや、ストリームがスループット制限を超えて抑制されたことが原因で、Kinesis Data Streams の put オペレーションが失敗することがあるためです。データが失われることがないように、単純な再試行を使用するなど、put オペレーションの再試行ポリシーを慎重に検討することをお勧めします。

• ステータスのログ記録は有益ですが、オプションです。

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