Redis Streams: xadd + xread

Esta ocasión mostraré un ejemplo de Redis Streams, con los metodos xadd y xread utilizando la librería Lettuce. 

Generar Mensajes
El Stream tendrá el nombre mensajes y cada mensaje enviado tendrá un UUID y un número indicando la cantidad de mensajes enviados:

    final var cliente = RedisClient.create("redis://localhost:6379");
    try (final var con = cliente.connect()) {
      final var cmd = con.sync();
      int i = 0;
      while (i < 500_000) {
        final var res =
            cmd.xadd("mensajes", Map.of(UUID.randomUUID().toString(), 
                                        String.valueOf(i)));
        i++;
        LOGGER.info("enviados: {} - id: {}", i, res);
        Thread.sleep(1_000);
      }
    }
    cliente.shutdown(); 

Recibir Mensajes
El cliente lee un mensaje a la vez y recuerda el último ID leido para preguntar por los nuevos mensajes y se bloquea 5 segundos esperando mensajes:

    final var cliente = RedisClient.create("redis://localhost:6379");
    String ultimo = "0";
    boolean leyendo = true;
    var cuentaMensajes = 0;
    try (final var con = cliente.connect()) {
      final var cmd = con.sync();
      while (leyendo) {
        final var mensajes = cmd.xread(
              XReadArgs.Builder.block(Duration.ofSeconds(5)).count(1),
              StreamOffset.from("mensajes", ultimo));
        cuentaMensajes += mensajes.size();
        LOGGER.info("leidos: {}", cuentaMensajes);
        for (final var m : mensajes) {
          LOGGER.info("{}", m.getId());
          ultimo = m.getId();
        }
      }
    }
    cliente.shutdown(); 

El código es bastante simple (quitando los extras como la pausa y los comentarios quedarian menos lineas) y cumple el objetivo de mostrar lo sencillo que es utilizar xadd y xread con Lettuce para Streams del poderoso Redis 💪.

El código queda disponible en GitHub 🙃.