Implementasikan produsen

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);
        }
    }
                    

Lihat rincian kode berikut:

• PutRecordAPIMengharapkan array byte, dan Anda harus mengonversi perdagangan ke JSON format. Baris kode tunggal ini melakukan operasi itu:

  
  byte[] bytes = trade.toJsonAsBytes();
                              

• Sebelum Anda dapat mengirim perdagangan, Anda membuat PutRecordRequest instance baru (disebut permintaan dalam kasus ini). Masing-masing request membutuhkan nama aliran, kunci partisi, dan gumpalan data.

  
  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();
                              

  Contoh menggunakan ticker saham sebagai kunci partisi, yang memetakan catatan ke pecahan tertentu. Dalam praktiknya, Anda harus memiliki ratusan atau ribuan kunci partisi per pecahan sehingga catatan tersebar merata di seluruh aliran Anda. Untuk informasi selengkapnya tentang cara menambahkan data ke aliran, lihatMenulis data ke Amazon Kinesis Data Streams.

  Sekarang request siap untuk mengirim ke klien (operasi put):

  
     kinesisClient.putRecord(request).get();
                              

• Pemeriksaan kesalahan dan pencatatan selalu merupakan tambahan yang berguna. Kode ini mencatat kondisi kesalahan:

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

  Tambahkan blok coba/tangkap di sekitar operasi: put

  
  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);
  }
                              

  Ini karena operasi put Kinesis Data Streams dapat gagal karena kesalahan jaringan, atau karena aliran data mencapai batas throughputnya dan terhambat. Disarankan agar Anda mempertimbangkan dengan cermat kebijakan coba ulang Anda untuk put operasi untuk menghindari kehilangan data, seperti menggunakan coba lagi.

• Pencatatan status sangat membantu tetapi opsional:

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