Enviar y recibir mensajes JSON con Spring Boot AMQP y RabbitMQ

Configurar una aplicación Hello-World de Spring Boot usando AMQP con RabbitMQ es bastante fácil si optas por la configuración básica, como lo hacen muchas otras guías. Sin embargo, la configuración no es tan sencilla cuando entras en la configuración de serialización y quieres hacer uso de anotaciones @RabbitListener para producir y consumir mensajes en formato JSON. Por lo tanto, comparto contigo un enfoque realmente simple pero más serio que esas aplicaciones hello-messaging, respaldado con un ejemplo práctico de la vida real.

Este artículo cubre:

• Cómo enviar/publicar objetos Java como mensajes JSON usando Spring Boot y RabbitTemplate de RabbitMQ.
• Cómo leer/consumir mensajes JSON como objetos Java usando Spring Boot y la anotación @RabbitListener de RabbitMQ.
• Cómo enviar y recibir objetos Java a través de RabbitMQ usando el serializador Java predeterminado.

Actualizado: Los ejemplos de código ahora usan Java 17 y Spring Boot 2.5.5

Video Tutorial (Versiones anteriores)

¿Prefieres la versión en video? Usa versiones anteriores de Java y Spring Boot, pero contiene no solo la explicación sobre cómo configurar la configuración para enviar y recibir mensajes JSON sino también algunos detalles adicionales sobre cómo funcionan los Message Converters. El video completo dura 29 minutos, pero aquí tienes los atajos a las diferentes secciones:

• 👉 Parte 1 Enviar y recibir mensajes usando AMQP con formato Java predeterminado
• 👉 Parte 2 Ejecutar la aplicación Spring Boot AMQP
• 👉 Parte 3 Spring Boot AMQP Message Converters en detalle
• 👉 Parte 4 Usar JSON Converters para serializar mensajes

Si prefieres el código más reciente (o no te gustan los videos) solo sigue leyendo para la guía completa actualizada.

Introducción

Como probablemente sepas si llegaste a este artículo, AMQP es una especificación, un protocolo que define mensajería y patrones para la comunicación entre sistemas. RabbitMQ es un software que implementa AMQP. Si quieres saber más sobre sistemas de mensajería y RabbitMQ en particular, consulta mi libro ya que dedica un capítulo completo a estos conceptos.

Objetivo

Dentro de este artículo configuraremos una aplicación Spring Boot que usa RabbitMQ para enviar mensajes a un exchange Topic formateado como JSON, y leerlos usando dos colas diferentes:

• La primera cola leerá el contenido como una clase genérica org.springframework.amqp.core.Message.
• La segunda cola transformará el mensaje JSON a una clase específica creada por nosotros, CustomMessage.

Consulta el diagrama a continuación para obtener ayuda visual.

También te mostraré una configuración alternativa para hacer que esto funcione sin formatear el mensaje a JSON, aunque hoy en día, este formato es una buena elección para la intercomunicación entre aplicaciones y sistemas que pueden diferir en lenguajes de programación, por ejemplo.

Te recomiendo que uses los ejemplos de código mientras sigues estas instrucciones. Solo clona el repositorio GitHub. ¡No olvides darle una estrella si te resulta útil!

Repositorio GitHub: Puedes encontrar todo el código fuente en GitHub: Spring Boot JSON AMQP messaging. Si te gusta, ¡dale una estrella!

Configurando el proyecto

pom.xml

Crear una aplicación Spring Boot simple es bastante sencillo, en mi caso uso Spring Boot 2.4 y Maven. Puedes ir al sitio web start.spring.io y seleccionar la versión y las dependencias incluidas. Lo único que necesitas hacer para usar AMQP con RabbitMQ y JSON en Spring Boot es incluir los starters correspondientes para AMQP y las bibliotecas Jackson. Consulta el listado a continuación.

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.thepracticaldeveloper</groupId>
    <artifactId>spring-boot-amqp-messaging</artifactId>
    <version>1.0.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>Spring Boot AMQP messaging</name>
    <description>Skeleton project for RabbitMQ JSON Configuration</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>17</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-json</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

Dependencias del proyecto en pom.xml

El mensaje

Enviaremos un mensaje muy simple sobre RabbitMQ, pero es lo suficientemente bueno para tener una idea de cómo se ve la serialización en diferentes formatos. Te presento el record CustomMessage:

package com.thepracticaldeveloper.rabbitmqconfig;

import java.io.Serializable;

import com.fasterxml.jackson.annotation.JsonProperty;

public record CustomMessage(@JsonProperty("text") String text,
                            @JsonProperty("priority") int priority,
                            @JsonProperty("secret") boolean secret)
        implements Serializable {
}

El Java Record CustomMessage

Como puedes ver, es un objeto simple donde ponemos algo de texto, un entero llamado priority, y si el mensaje es secret o no. No construiremos ninguna lógica de negocio sobre estos tres campos, sin embargo, es bueno agregarlos para trabajar con contenido que esté cerca de escenarios de la vida real. Solo hay dos observaciones aquí:

• Necesitamos usar @JsonProperty en la declaración de nuestro Java record para instruir a las bibliotecas JSON a usar los nombres de campo correspondientes al serializar y deserializar.
• Necesitamos marcar nuestro record como serializable para que las bibliotecas Jackson funcionen bien.

Enviando mensajes

Nuestro ejemplo práctico usará un topic de tipo Exchange para enviar los objetos CustomMessage:

package com.thepracticaldeveloper.rabbitmqconfig;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.Random;

@Service
public class CustomMessageSender {

    private static final Logger log = LoggerFactory.getLogger(CustomMessageSender.class);

    private final RabbitTemplate rabbitTemplate;

    public CustomMessageSender(final RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @Scheduled(fixedDelay = 3000L)
    public void sendMessage() {
        final var message = new CustomMessage("Hello there!", new Random().nextInt(50), false);
        log.info("Sending message...");
        rabbitTemplate.convertAndSend(MessagingApplication.EXCHANGE_NAME, MessagingApplication.ROUTING_KEY, message);
    }
}

El servicio CustomMessage Sender

Algunos comentarios sobre este servicio:

• La anotación @Scheduled se usa para programar un método para que se ejecute en un momento dado, o con una frecuencia dada. En este caso, este método se ejecutará cada 3 segundos (hasta el fin del mundo o hasta que detengas la aplicación, lo que ocurra primero). Consulta los Spring docs si quieres saber más sobre cómo funciona esto.
• RabbitTemplate se usa para convertir y enviar un mensaje usando RabbitMQ. Es una clase helper, como muchas otras clases Template existentes en Spring (como JdbcTemplate, RestTemplate, etc.). Spring Boot AMQP proporciona un RabbitTemplate predeterminado, pero necesitaremos ajustarlo un poco para usar JSON para serializar los mensajes.
• El exchange y la routing key son conceptos AMQP. Hay múltiples tipos de exchange para diferentes casos de uso, y las routing keys ayudan a RabbitMQ a saber dónde poner los mensajes para ser consumidos. Te recomiendo conseguir mi libro si quieres profundizar en estos conceptos y aprender cómo RabbitMQ puede soportar una arquitectura de microservicios. Como resumen, estamos enviando mensajes a través de un canal dado (exchange), y la routing key nos permite filtrar para quién son los generados.

Veremos más tarde dónde declarar el Exchange topic, la Queue, y el binding que usa la Routing Key.

Recibiendo mensajes

Necesitamos construir el otro lado de la comunicación también. Para esta publicación, he creado los receivers en la misma aplicación para mejor logística (tener un único repositorio), pero no deberías tener problemas para replicar esta configuración con dos aplicaciones diferentes si quieres. De hecho, podrías copiar esta aplicación Spring Boot dos veces, eliminar el productor en una de las copias y los consumidores en la otra, y ejecutar ambas. Consulta el consumidor RabbitMQ:

package com.thepracticaldeveloper.rabbitmqconfig;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
public class CustomMessageListener {

    private static final Logger log = LoggerFactory.getLogger(CustomMessageListener.class);

    @RabbitListener(queues = MessagingApplication.QUEUE_GENERIC_NAME)
    public void receiveMessage(final Message message) {
        log.info("Received message as a generic AMQP 'Message' wrapper: {}", message.toString());
    }

    @RabbitListener(queues = MessagingApplication.QUEUE_SPECIFIC_NAME)
    public void receiveMessage(final CustomMessage customMessage) {
        log.info("Received message and deserialized to 'CustomMessage': {}", customMessage.toString());
    }
}

Un consumidor RabbitMQ usando anotaciones @RabbitListener

Aquí puedes ver nuestro objetivo: la idea es mostrar al mismo tiempo cómo el mismo mensaje, producido por la clase CustomMessageSender, puede ser recibido a través de dos colas diferentes en dos formatos diferentes. La primera cola (GENERIC) recibe el mensaje como una clase AMQP Message genérica, por lo que no hay conversión implícita. Alerta de spoiler: cuando registramos el mensaje deberíamos ver un string JSON en el payload ya que no ha sido convertido a una clase Java.

Por otro lado, la cola SPECIFIC declara un argumento CustomMessage en el método anotado con RabbitListener. Esto activará una lógica dentro de Spring para encontrar un conversor de JSON a esa clase específica. Aunque esta parte debería ser fácil de configurar, a veces es la parte en la que nosotros los desarrolladores luchamos más. En este caso es fácil adivinar que el texto registrado debería ser la implementación de toString() del CustomMessage (que es en sí misma la implementación predeterminada para records).

Configuración

Esta es la parte más interesante ya que es lo que hace que todo funcione junto. Primero, encontrarás algunas anotaciones en la clase MessagingApplication:

@SpringBootApplication
@EnableScheduling
public class MessagingApplication {

Anotaciones en la clase principal de Spring Boot

Además de la que hace que Spring Boot funcione (@SpringBootApplication), necesitamos agregar la anotación @EnableScheduling para usar el método anotado con @Scheduled para enviar mensajes que vimos anteriormente. Como nota al margen, no necesitas agregar la anotación @EnableRabbit de Spring ya que RabbitMQ se habilitará automáticamente solo porque agregaste la dependencia. Así es como funciona la autoconfiguración de Spring Boot (también detallado en mi libro si estás interesado).

Debajo de estas anotaciones, encontrarás algunas constantes y, junto a ellas, la configuración de Exchange / Queue:

@Bean
public TopicExchange appExchange() {
    return new TopicExchange(EXCHANGE_NAME);
}

@Bean
public Queue appQueueGeneric() {
    return new Queue(QUEUE_GENERIC_NAME);
}

@Bean
public Queue appQueueSpecific() {
    return new Queue(QUEUE_SPECIFIC_NAME);
}

@Bean
public Binding declareBindingGeneric() {
    return BindingBuilder.bind(appQueueGeneric()).to(appExchange()).with(ROUTING_KEY);
}

@Bean
public Binding declareBindingSpecific() {
    return BindingBuilder.bind(appQueueSpecific()).to(appExchange()).with(ROUTING_KEY);
}

Declaración de objetos RabbitMQ vía declaración de bean

Estos son el topic exchange, ambas colas y los bindings entre ellos. Ambas colas están vinculadas al mismo exchange con la misma routing key, por lo que ambas obtendrán una copia de cada mensaje que se produce. Esta es solo una configuración de ejemplo, podrías usar muchas más dependiendo de tus necesidades.

En la parte inferior, en la misma clase, podemos encontrar la configuración relacionada con producir y consumir mensajes JSON:

@Bean
public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
    final var rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMessageConverter(producerJackson2MessageConverter());
    return rabbitTemplate;
}

@Bean
public Jackson2JsonMessageConverter producerJackson2MessageConverter() {
    return new Jackson2JsonMessageConverter();
}

Configuración de RabbitTemplate para serialización JSON

Con esa configuración, anulamos el bean RabbitTemplate predeterminado creado por Spring Boot con otro que usará un convertidor de mensajes de tipo Jackson2JsonMessageConverter. Recuerda: esta configuración solo afecta a los mensajes enviados vía RabbitTemplate.

El bean Jackson2JsonMessageConverter también se encargará de deserializar los mensajes JSON a clases Java, usando un ObjectMapper predeterminado de Jackson.

Ten en cuenta que, con esta configuración, no necesitas implementar la interfaz RabbitListenerConfigurer ni usar un MappingJackson2MessageConverter como era necesario en una versión anterior de esta guía.

Ejecutando la aplicación

Primero, necesitas asegurarte de que el servidor RabbitMQ esté ejecutándose. En el repositorio GitHub, hay un archivo docker-compose.yml que puedes usar si tienes Docker instalado. Solo ejecuta:

docker-compose up -d

Además, si quieres usar una configuración diferente a la predeterminada para conectarte a RabbitMQ, asegúrate de incluir esos cambios dentro del archivo application.properties (número de puerto, autenticación, etc.). No necesitas hacer esto si usas la configuración predeterminada como está establecida en esta base de código de ejemplo (a menos que tu IP de host de Docker no sea localhost, entonces podrías necesitar ajustar la configuración).

Puedes ejecutar la clase de aplicación desde tu IDE preferido o ejecutando mvn spring-boot:run desde la línea de comandos (también puedes usar el wrapper de Maven embebido para eso). Consulta la salida de consola a continuación:

INFO 23562 --- [   scheduling-1] c.t.rabbitmqconfig.CustomMessageSender   : Sending message...
INFO 23562 --- [ntContainer#0-1] c.t.r.CustomMessageListener              : Received message as a generic AMQP 'Message' wrapper: (Body:'{"text":"Hello there!","priority":26,"secret":false}' MessageProperties [headers={__TypeId__=com.thepracticaldeveloper.rabbitmqconfig.CustomMessage}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=appExchange, receivedRoutingKey=messages.key, deliveryTag=3, consumerTag=amq.ctag-Q2Y3DUjFV_YHDPjQPOLuiQ, consumerQueue=appGenericQueue])
INFO 23562 --- [ntContainer#1-1] c.t.r.CustomMessageListener              : Received message and deserialized to 'CustomMessage': CustomMessage[text=Hello there!, priority=26, secret=false]

Ejecutando la aplicación con serialización JSON

¡Funciona como se esperaba! Vamos a describir qué sucede aquí.

• El productor (CustomMessageSender) envía un mensaje vía el Exchange topic.
• El método listener que consume objetos Message genéricos obtiene todos los headers de RabbitMQ y el cuerpo del payload, que se deserializa como un string. Dado que el mensaje se produjo como JSON, vemos esa representación allí.
• El otro listener declara un argumento CustomMessage, por lo que Spring simplemente toma el cuerpo del payload y usa la representación JSON para deserializarlo a Java. Tiene éxito, por lo que podemos trabajar directamente con el objeto. En nuestro caso, lo imprimimos como un String, y eso es lo que vemos en la consola. Es el formato toString() estándar para un Java record.

Cambiando de formato JSON a serialización predeterminada

Puedes comentar las líneas que configuran los convertidores de mensajes. Ese es el contenido completo de la sección de configuración JSON anterior. Lo que verás en ese caso es la configuración predeterminada para Spring Boot, que usa una serialización Java para codificar mensajes.

INFO 25142 --- [   scheduling-1] c.t.rabbitmqconfig.CustomMessageSender   : Sending message...
INFO 25142 --- [ntContainer#0-1] c.t.r.CustomMessageListener              : Received message as a generic AMQP 'Message' wrapper: (Body:'[B@7389c33f(byte[143])' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=appExchange, receivedRoutingKey=messages.key, deliveryTag=7, consumerTag=amq.ctag-9WtY0LRLuKgRrIzv3NiAbg, consumerQueue=appGenericQueue])
INFO 25142 --- [ntContainer#1-1] c.t.r.CustomMessageListener              : Received message and deserialized to 'CustomMessage': CustomMessage[text=Hello there!, priority=2, secret=false]

Ahora, como puedes ver en los headers del objeto Message, el formato del contenido (contentType) es application/x-java-serialized-object. Este es solo otro enfoque para la serialización, que podría resultar no ser el mejor si estás usando, por ejemplo, diferentes tipos de aplicaciones (no Java) para recibir mensajes. Por otro lado, el payload es más pequeño, lo que puede llevar a un mejor rendimiento si transfieres muchos datos.

En cualquier caso, la serialización y deserialización funciona perfectamente. Normalmente, optarías por la segunda opción y abstraerías el formato de serialización de tus listeners, como en el listener CustomMessage, que se comporta exactamente igual que antes.

Más información

Espero que hayas encontrado útil esta guía. Si estás interesado en usar RabbitMQ en una Arquitectura de Microservicios para soportar una Arquitectura Orientada a Eventos, y algunos otros temas relacionados como Service Discovery, Routing, Comunicación Async vs. Sync, etc., no te pierdas mi libro Learn Microservices with Spring Boot. Te guiará paso a paso para obtener experiencia real, mirando también las desventajas de este tipo de configuración.

Consigue el libro: Disponible en Amazon