Implementar el productor

Consulte el siguiente desglose del código:

• La API de PutRecord espera una matriz de bytes, y debe convertir el trade a formato JSON. Esta única línea de código realiza esa operación:

  byte[] bytes = trade.toJsonAsBytes();

• Antes de poder enviar la transacción, debe crear una nueva instancia de PutRecordRequest (denominada putRecord en este caso):

  PutRecordRequest putRecord = new PutRecordRequest();

  Cada llamada a PutRecord requiere el nombre de la secuencia, la clave de partición y el blob de datos. El siguiente código rellenará estos campos en el objeto putRecord utilizando sus métodos setXxxx():

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

  El ejemplo utiliza un ticker como clave de partición, que asigna el registro a una partición específica. En la práctica, debería tener cientos o miles de claves de partición por fragmento, de forma que los registros se dispersen de forma uniforme en toda la secuencia. Para obtener más información acerca de cómo agregar datos a una secuencia, consulte Agregar datos a un flujo.

  Ahora putRecord estará listo para el envío al cliente (la operación put):

  kinesisClient.putRecord(putRecord);

• Siempre es útil agregar funciones de comprobación y registro de errores. Este código registra las condiciones de error:

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

  Añada el try/catch bloque alrededor de la put operación:

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

  Esto se debe a que una operación put de Kinesis Data Streams puede fallar debido a un error de red o debido a que el flujo alcanza sus límites de rendimiento y se ve limitado. Le recomendamos comprobar detalladamente su política de reintentos para las operaciones de put de modo que evite pérdida de datos, por ejemplo, al utilizar un reintento.

• El registro de estado resulta útil, pero es opcional:

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