RabbitMQ, Patrones de Mensajería y Comparativa Estratégica con Kafka

@Configuration
class RabbitTopologia {

    @Bean
    TopicExchange tareasExchange() {
        return new TopicExchange("tareas");                      // durable por defecto
    }

    @Bean
    Queue notificacionesQueue() {
        return QueueBuilder.durable("tareas.notificaciones").quorum().build();
    }

    @Bean
    Queue auditoriaQueue() {
        return QueueBuilder.durable("tareas.auditoria").quorum().build();
    }

    @Bean
    Binding soloCompletadas(Queue notificacionesQueue, TopicExchange tareasExchange) {
        // tareas.<proyecto>.completada  ->  notificaciones
        return BindingBuilder.bind(notificacionesQueue).to(tareasExchange).with("tareas.*.completada");
    }

    @Bean
    Binding todosLosEventos(Queue auditoriaQueue, TopicExchange tareasExchange) {
        return BindingBuilder.bind(auditoriaQueue).to(tareasExchange).with("tareas.#");
    }
}

// application.yml
// spring:
//   rabbitmq:
//     listener:
//       simple:
//         acknowledge-mode: manual
//         prefetch: 10            # como máximo 10 mensajes sin confirmar por consumidor
//         concurrency: 2
//         max-concurrency: 8

@RabbitListener(queues = "tareas.notificaciones")
void alRecibir(TareaEvento evento, Channel canal,
               @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
    try {
        notificador.avisar(evento);
        canal.basicAck(tag, false);                    // procesado: RabbitMQ lo borra
    } catch (ServicioCaidoException e) {
        canal.basicNack(tag, false, true);             // transitorio: vuelve a la cola
    } catch (Exception e) {
        canal.basicNack(tag, false, false);            // permanente: al DLX si está configurado
    }
}

@Bean
Queue principal() {
    return QueueBuilder.durable("pagos.procesar")
        .deadLetterExchange("pagos.dlx")
        .deadLetterRoutingKey("espera")
        .build();
}

@Bean
Queue espera() {                                      // sin consumidores: solo espera
    return QueueBuilder.durable("pagos.espera")
        .ttl(30_000)                                  // 30 s
        .deadLetterExchange("pagos")                  // al caducar, vuelve al exchange principal
        .deadLetterRoutingKey("procesar")
        .build();
}

@Bean
Queue aparcados() {
    return QueueBuilder.durable("pagos.aparcados").build();
}

@RabbitListener(queues = "pagos.procesar")
void procesar(Message mensaje, Pago pago) {
    long intentos = Optional.ofNullable(mensaje.getMessageProperties().getXDeathHeader())
        .map(d -> d.stream().mapToLong(x -> (Long) x.get("count")).sum()).orElse(0L);
    if (intentos >= 3) {
        rabbit.convertAndSend("", "pagos.aparcados", pago);  // revisión manual
        return;
    }
    pasarela.cobrar(pago);                            // si lanza excepción -> DLX -> espera 30 s
}

                         RabbitMQ                          Kafka
Modelo                   cola con enrutamiento             log particionado
Tras consumir            el mensaje se borra (ack)         se conserva según retención
Releer el pasado         no                                sí (mover offset)
Enrutamiento             exchanges direct/topic/fanout     por topic y clave de partición
Orden                    por cola (con 1 consumidor)       por partición
Reintento por mensaje    nack, DLX, TTL                    error handler, topics de reintento, DLT
Rendimiento típico       decenas de miles msg/s            cientos de miles a millones msg/s
Encaja con               colas de trabajo, RPC, tareas     event streaming, event sourcing, CDC

@Bean
MessageConverter jsonConverter(ObjectMapper mapper) {
    return new Jackson2JsonMessageConverter(mapper);     // Boot lo usa en template y listeners
}

@Service
class PublicadorTareas {
    private final RabbitTemplate rabbit;
    PublicadorTareas(RabbitTemplate rabbit) { this.rabbit = rabbit; }

    void tareaCompletada(Tarea t) {
        var evento = new TareaEvento(t.id(), "COMPLETADA", Instant.now());
        rabbit.convertAndSend("tareas", "tareas." + t.proyecto() + ".completada", evento);
    }
}

// application.yml
// spring:
//   rabbitmq:
//     publisher-confirm-type: correlated
//     listener:
//       simple:
//         default-requeue-rejected: false     # agotados los reintentos -> DLX
//         retry:
//           enabled: true
//           max-attempts: 4
//           initial-interval: 1s
//           multiplier: 2