08 - Eventos de aplicación y mensajería

La llamada a la API pública de otro módulo es comunicación síncrona: quien llama necesita el resultado para continuar. Los eventos de aplicación permiten la comunicación inversa: un módulo anuncia que algo ha ocurrido y otros módulos reaccionan, sin que el emisor los conozca.

Un evento encaja cuando la operación auxiliar no forma parte del flujo principal de la tarea:

El criterio es la consistencia: si la acción auxiliar no ocurre, la operación principal sigue siendo correcta y no deja el sistema en un estado inválido, entonces puede ser un evento. Si el sistema queda inconsistente sin ella, debe estar en el flujo principal.

Spring proporciona un mecanismo de publicación con ApplicationEventPublisher. El evento es un objeto inmutable que describe lo ocurrido; el consumidor declara un método listener:

// En el módulo de pedidos
class OrderServiceImpl implements OrderService {
    private final ApplicationEventPublisher events;

    public OrderSummaryDto cancel(CancelOrderDto request) {
        Order order = // ... regla de negocio ...
        order.cancel();
        events.publishEvent(new OrderCancelled(order.id()));
        return mapper.toSummary(order);
    }
}

// En el módulo de notificaciones
@Component
class NotificationListener {
    @EventListener
    void on(OrderCancelled event) {
        // Enviar correo de confirmación.
    }
}

El emisor conoce el evento, no a los consumidores: el módulo de pedidos no importa ninguna clase del módulo de notificaciones. Añadir un nuevo consumidor no cambia al emisor.

La publicación es síncrona por defecto: el listener se ejecuta en la misma llamada y dentro de la transacción de quien publica. Mantener una única transacción conserva la consistencia, pero amplía su alcance: si el listener falla, falla la operación principal.

@Async
@TransactionalEventListener
void on(OrderCancelled event) {
    // Se ejecuta tras el commit, fuera de la transacción del emisor.
}

Con @TransactionalEventListener y @Async el trabajo secundario se ejecuta después del commit y desacoplado del flujo. El coste es la fiabilidad: si el proceso cae entre el commit y la ejecución, la publicación se pierde.

Spring Modulith ofrece @ApplicationModuleListener, abreviatura de la combinación habitual para eventos entre módulos: asíncrona y transaccional. Además, su registro de publicaciones guarda cada publicación en la base de datos y la marca como completada o pendiente; si el listener falla, la publicación queda registrada para reintentarla, resolviendo el riesgo de pérdida del modo asíncrono.

Los eventos de aplicación viven dentro del proceso. Cuando el consumidor está en otra aplicación, la publicación debe atravesar la red: un broker de mensajes recibe el mensaje del emisor y lo entrega a los consumidores registrados.

RabbitMQ y Kafka son los brokers más habituales:

El broker introduce requisitos nuevos: serialización del mensaje, versiones del contrato, entrega al menos una vez (los consumidores deben tolerar duplicados) y supervisión de mensajes sin procesar.

Kafka organiza el flujo como un log distribuido: cada topic se divide en particiones, y cada mensaje queda registrado en su partición hasta que la retención lo expira. El orden se garantiza dentro de la partición; la clave del mensaje decide la partición, así que los mensajes con la misma clave se procesan en orden.

Los consumidores avanzan leyendo un offset por partición. Un consumer group reparte las particiones entre sus miembros: el paralelismo de consumo es el número de particiones. Como el offset es la posición de cada grupo, un grupo nuevo o suscrito tarde puede leer el histórico completo del topic si la retención lo permite.

// Publicar (spring-kafka)
kafkaTemplate.send("order-events", String.valueOf(order.id()), event);

// Consumir
@KafkaListener(topics = "order-events", groupId = "notifications")
void onOrderCancelled(OrderCancelled event) {
    // Reacción secundaria.
}

services:
  kafka:
    image: apache/kafka
    ports:
      - "9092:9092"

RabbitMQ usa el modelo de colas con routing: el productor envía a un exchange, que reparte los mensajes a las colas vinculadas según la clave de enrutado. Los tipos habituales: fanout (difusión a todas), direct (coincidencia exacta) y topic (coincidencia por patrón). Al confirmar el consumo, el mensaje se retira de la cola: lo publicado no queda como un log que se pueda volver a leer.

// Publicar (spring-amqp)
rabbitTemplate.convertAndSend("order-exchange", "order.cancelled", event);

// Consumir
@RabbitListener(queues = "order-notifications")
void onOrderCancelled(OrderCancelled event) {
    // Reacción secundaria.
}

services:
  rabbit:
    image: rabbitmq:4-management
    ports:
      - "5672:5672"
      - "15672:15672"

Spring Modulith conecta los eventos internos con los brokers mediante la anotación @Externalized, que marca qué eventos salen del proceso y su destino:

@Externalized("order-events::${spring.application.name}")
public record OrderCancelled(long orderId) {}

Con la integración de Kafka o AMQP (RabbitMQ), el evento se publica en el broker al completarse la transacción, apoyándose en el registro de publicaciones para los reintentos. La aplicación externa consume el mensaje con las herramientas del broker, sin conocer la aplicación emisora. Las integraciones concretas están descritas en Soporte de ejecución.

• Probar que la operación principal publica el evento esperado con los datos correctos.
• Probar cada listener de forma aislada: recibe el evento y realiza su acción auxiliar.
• Probar la integración emisor-consumidor con el contexto de módulos; para listeners asíncronos, esperar la reacción con la API Scenario.
• Probar la fiabilidad: un listener que falla deja la publicación pendiente y puede reintentarse.
• Probar la externalización de punta a punta contra el broker real en un contenedor efímero (Testcontainers): publicar el evento y consumirlo con la integración del broker.

@ApplicationModuleTest
class OrderCancelledScenarioTest {
    @Test
    void notificationReactsToCancellation(Scenario scenario) {
        scenario.mutate(() -> orderService.cancel(new CancelOrderDto(42L)))
                .andWaitForEventOfType(NotificationSent.class)
                .toArriveAndVerify(event ->
                        assertThat(event.orderId()).isEqualTo(42L));
    }
}

La publicación del estímulo se ejecuta en su propia transacción y la espera del evento resultante sustituye a las pausas arbitrarias de los tests con hilos asíncronos.