Introducción a Apache Pulsar en Spring Boot

En esta entrada vamos a ver una introducción a Apache Pulsar en Spring Boot, en el que veremos mediante un ejemplo su uso y funcionamiento.

En artículos anteriores hemos visto como hacer uso de Kafka para hacer uso de eventos entre consumidores y productores en una arquitectura de microservicios. Hoy vamos a ver como podemos aplicar estas arquitecturas viendo una introducción a Apache Pulsar en Spring Boot.

¿Qué es Apache Pulsar?

Apache Pulsar es un sistema distribuido de mensajería Publish/Subscribe creado por Yahoo.

Fue puesto bajo licencia Open Source en 2016.

Arquitectura de Apache Pulsar

Apache Pulsar es una solución multi tenant de alto rendimiento para mensajería de servidor a servidor. Tiene un Zookeeper incorporado para la configuración y administración y se encuentra compuesto por un conjunto de brokers y bookies. Los bookies son de Apache BookKeeper, la cual proporciona un almacenamiento para los mensajes hasta que se consuman.

En un clúster de Apache Pulsar tendremos:

• Múltiples brookers para manejar los mensajes entrantes y gestionar los mensajes hacia los consumidores.
• Apache BookKeeper para la persistencia de los mensajes
• Apache Zookeeper para la configuración del clúster.

Características principales de Apache Pulsar

Vamos a echar un ojo a las principales características de Apache Pulsar:

• Compatibilidad con replicación geográfica.
• Baja latencia
• Bookkeeper para garantizar la entrega de los mensajes
• Escalable a un elevado número de topics.
• Multiples modos de suscripción.
• Soporte para múltiples clúster.

Vamos a intentar entrar en más detalle para explicar estas características.

Compatibilidad con replicación Geográfica en Apache Pulsar

Apache Pulsar nos va a permitir activar la replicación entre diferentes clúster que se encuentran en diferentes regiones geográficas. La replicación de los mensajes se realiza en real time, permitiendo una gran tolerancia al fallo ya que en caso de caídas o errores en la red, los mensajes se guardan en Bookkeeeper.

El uso de replicación geográfica que nos proporciona Pulsar nos va a permitir desplegar en diferentes regiones, de manera que podemos evitar hacer uso de las propiedades multi región que nos ofrecen algunos clouds.

Soporte para múltiples clúster

Pulsar nos ofrece la posibilidad de despliegue en diferentes entornos e infraestructuras, lo que nos ayudará a realizar despliegues en diferentes clouds o en entornos on-premise.

Durabilidad y consistencia

Una de las características de Pulsar es la garantía de que no hay perdida de información una vez se ha recibido el ACK. Pulsar hace uso de Write Ahead Log para evitar la perdida de información incluso fallo del servidor, escribe logs y guarda una copia en memoria para asegurar la consistencia de la información.

Además la información es replicada a múltiples bookies, con lo que se garantiza la consistencia y la no pérdida de información.

Tipos de suscripción y modelado del mensaje

El sistema de mensajería que nos ofrece Pulsar acepta dos tipos, colas y Publisher/Subscriber.

Por un lado Publisher/Subscriber garantiza la antrega de los mensajes a todos los consumidores que se encuentran suscritos y por otro lado tenemos un sistema de colas en el que la comunicación es punto a punto.

Pulsar permite a los consumidores elegir que sistema de consumo va a hacer uso.

En Apache Pulsar encontramos 4 tipos diferentes de Suscripción:

• Exclusive: Solo un consumidor se suscribe a un topic de una partición. Esto no escala bien porque solo habrá un consumidor consumiendo mensajes.
• Failover: Multiples consumidores se pueden suscribir pero solo hay uno consumiendo al mismo tiempo.
• Shared: Multiples consumidores consumen al mismo tiempo. Los mensajes se entregan a los consumidores con una política de Round Robin.
• Key_Shared: Este tipo garantiza el orden y además permite que se conecten varios consumidores al mismo tiempo. Con este tipo de suscripción hay que informar de la clave.

Ejemplo con Pulsar en Spring Boot

A continuación vamos a ver y a configurar un ejemplo haciendo uso de un único clúster de Apache Pulsar y a conectarlo a través de su API con una aplicación Spring Boot.

En nuestra aplicación enviaremos un mensaje desde un consumidor a un productor haciendo uso de Publisher/Subscriber.

Actualmente no existe ningún starter oficial de Spring para apache Pulsar por lo que hacemos uso de la librería nativa de Java.

Para poder desplegar y levantar un Apache Pulsar, podemos hacerlo de dos maneras diferentes, a través de docker o través de una instalación.

El ejemplo completo lo puedes descargar de aquí.

Ejecutar a través de docker Apache Pulsar

Para ejecutar con Docker podemos hacer uso del siguiente comando.

$ docker run -it -p 6650:6650 -p 8080:8080 --mount source=pulsardata,target=/pulsar/data --mount source=pulsarconf,target=/pulsar/conf apachepulsar/pulsar:2.9.1 bin/pulsar standalone

Y una vez arrancado podremos usar cualquier de las siguientes urls:

• pulsar://localhost:6650
• http://localhost:8080

Instalar Apache Pulsar con los binarios

Otra manera de poder tener una instancia de Apache Pulsar es partir de los binarios:

$ wget https://archive.apache.org/dist/pulsar/pulsar-2.9.1/apache-pulsar-2.9.1-bin.tar.gz

y a continuación ir dentro de la carpeta:

$ tar xvfz apache-pulsar-2.9.1-bin.tar.gz

$ cd apache-pulsar-2.9.1

Dependencias Maven para una aplicación Spring Boot con Pulsar

Pulsar nos ofrece la siguiente dependencia para poder conectarnos y configurar el acceso al clúster:

        <dependency>
            <groupId>org.apache.pulsar</groupId>
            <artifactId>pulsar-client</artifactId>
            <version>2.9.1</version>
        </dependency>

Configuración del Cliente de Pulsar en Java

Para poder crear tanto un productor como un consumidor para hacer uso de Pulsar es necesario primero crear un cliente. Para ello hacemos uso de la clase PulsarClient que nos proporciona un builder para crear el cliente:

PulsarClient pulsarClient= PulsarClient.builder()
  .serviceUrl(SERVICE_URL)
  .build();

Este cliente nos va a permitir conectar tanto un productor como un consumidor.

Configuración del Productor de Apache Pulsar

Una vez tenemos nuestro cliente configurado, creamos un Productor que se conectará a nuestro cliente:

Producer<byte[]> producer = client.newProducer()
  .topic(TOPIC_NAME)
  .compressionType(CompressionType.LZ4)
  .create();

Nuestro productor se le va a añadir el nombre del TOPIC al que se tiene que suscribir.

Configuración del Consumidor en Apache Pulsar

Al igual que hemos configurado nuestro Productor, vamos a conectar un Consumidor al cliente de Pulsar que hemos creado anteriormente:

Consumer<byte[]> consumer = client.newConsumer()
  .topic(TOPIC_NAME)
  .subscriptionType(SubscriptionType.Shared)
  .subscriptionName(SUBSCRIPTION_NAME)
  .subscribe();

Nuestro consumidor lo hemos configurado con un tipo de suscripción shared, en el que le indicamos que múltiples consumidores se adhieren a la misma suscripción y pueden consumir mensajes.

Una vez que nuestro consumidor y nuestro productor han sido configurados tan solo queda producir y recibir los mensajes.

Producir mensajes con Apache Pulsar

Nuestro productor va estar en un bucle durante 10 iteracciones y enviará el número en el que se encuentra:

  public void producer() throws PulsarClientException {
    Producer<byte[]> producer = client.newProducer()
        .topic(TOPIC_NAME)
        .compressionType(CompressionType.LZ4)
        .create();

    IntStream.range(1, 5).forEach(i -> {
      String content = String.format("message number: %d", i);

      // Send each message and log message content and ID when successfully received
      try {
        producer.send(content.getBytes());
      } catch (PulsarClientException e) {
        log.error("Error sending mesasage");
        e.printStackTrace();
      }

      System.out.println("Published message '"+content );

    });

    producer.close();

  }

Consumidor mensajes con Apache Pulsar

Nuestro consumidor se encargar de recibir los mensajes, para ello entrará en un bucle infinito:

  public void conumser() throws PulsarClientException {
    Consumer<byte[]> consumer = client.newConsumer()
        .topic(TOPIC_NAME)
        .subscriptionType(SubscriptionType.Shared)
        .subscriptionName(SUBSCRIPTION_NAME)
        .subscribe();

    do {
      Message<byte[]> msg = consumer.receive();

      String content = new String(msg.getData());
      System.out.println("Received message '" + content);

      consumer.acknowledge(msg);
    } while (true);

  }

Conclusión

En este artículo sobre introducción a Apache Pulsar en Spring Boot hemos visto la arquitectura así como las principales características de Pulsar, siendo una opción válida para una arquitectura de microservicios en la que tengamos que manejar eventos.

Si quieres ver el ejemplo completo puedes descargarte el código de aquí.