Implémenter le producteur

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

Reportez-vous à la ventilation de code suivante :

• Le PutRecord API attend un tableau d'octets, et vous devez convertir le trade en JSON format. La seule ligne de code suivante effectue cette opération :

  
  byte[] bytes = trade.toJsonAsBytes();
                              

• Avant d’envoyer l'opération, vous créez une nouvelle instance PutRecordRequest (appelée request dans le cas suivant) : Chaque appel de request requiert le nom du flux, la clé de partition et un blob de données.

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

  L'exemple utilise un ticker boursier comme clé de partition, qui fait correspondre l'enregistrement à une partition spécifique. En pratique, vous devez avoir des centaines ou des milliers de clés de partition par partition afin de répartir les enregistrements de façon égale dans votre flux. Pour plus d'informations sur la façon d'ajouter des données à un flux, consultez la pageÉcrire des données dans Amazon Kinesis Data Streams.

  Maintenant request est prêt à envoyer au client (l'opération put) :

  
     kinesisClient.putRecord(request).get();
                              

• La vérification des erreurs et leur consignation sont toujours très utiles. Le code suivant consigne les conditions d'erreur :

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

  Ajoutez le bloc try/catch autour de l'opération 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);
  }
                              

  Une opération put de Kinesis Data Streams peut échouer en raison d’une erreur réseau ou si le flux de données atteint ses limites de débit. Il est recommandé d'examiner attentivement votre politique de nouvelle tentative pour les put opérations afin d'éviter toute perte de données, par exemple en utilisant une nouvelle tentative.

• La journalisation d'état est très utile, mais elle est facultative :

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