MQTT con Spring Boot y Spring Integration

Spring Boot no tiene un cliente MQTT propio, pero Spring Integration sí: el módulo spring-integration-mqtt envuelve Eclipse Paho en channel adapters, de modo que los mensajes MQTT entrantes se convierten en objetos Message de Spring en un canal, y todo lo que envías a un canal de salida se publica en el broker. En este tutorial construyes un pequeño servicio que se suscribe, publica a través de una interfaz gateway y se puede probar de extremo a extremo desde el navegador.

Dependencias

La gestión de dependencias de Spring Boot cubre la versión de Spring Integration, así que no tienes que fijarla tú:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-mqtt</artifactId>
</dependency>

El módulo incluye el cliente Paho mqttv3 para MQTT 3.1.1. Si quieres los adapters de MQTT 5, añade también org.eclipse.paho:org.eclipse.paho.mqttv5.client. Los ejemplos usan Spring Boot 3 o superior y Java 17.

Spring Integration es la opción adecuada cuando MQTT es uno de varios puntos de integración y quieres canales, transformadores y gestión de errores en un mismo modelo. Si tu servicio solo necesita un cliente y quieres reactive streams o control total de MQTT 5, el HiveMQ MQTT Client envuelto en un bean de Spring normal es una alternativa más ligera.

Propiedades de configuración

Mantén la configuración del broker fuera del código. Un pequeño record enlazado con @ConfigurationProperties la mantiene tipada:

mqtt:
  url: ssl://broker.connectmq.com:8883
  username: ${MQTT_USER}
  password: ${MQTT_PASS}
  client-id: orders-service-${random.uuid}
  inbound-topic: testmqtt/spring/in/#
  outbound-topic: testmqtt/spring/out

@ConfigurationProperties(prefix = "mqtt")
public record MqttProperties(String url, String username, String password,
                             String clientId, String inboundTopic, String outboundTopic) {}

Actívalo con @EnableConfigurationProperties(MqttProperties.class) en una clase de configuración. Usa ssl:// con el puerto 8883 para TLS y tcp:// con el 1883 para conexiones en claro durante el desarrollo local.

La client factory

Todos los adapters obtienen sus conexiones de una MqttPahoClientFactory. Aquí configuras una sola vez TLS, credenciales, keep alive y el comportamiento de reconexión:

@Configuration
@EnableConfigurationProperties(MqttProperties.class)
@IntegrationComponentScan
public class MqttConfig {

    @Bean
    public MqttPahoClientFactory mqttClientFactory(MqttProperties props) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[] { props.url() });
        options.setUserName(props.username());
        options.setPassword(props.password().toCharArray());
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        options.setKeepAliveInterval(30);
        options.setConnectionTimeout(10);

        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(options);
        return factory;
    }
}

MqttConnectOptions es la clase normal de Paho, así que tienes disponibles todas sus opciones: Last Will, un SSLSocketFactory personalizado, la versión de MQTT. Consulta MQTT en Java con Eclipse Paho para los detalles de cada opción.

Entrada: suscribirse con un message-driven adapter

MqttPahoMessageDrivenChannelAdapter se suscribe a uno o varios filtros de topic y envía cada mensaje recibido a un canal de salida. Después, un @ServiceActivator lo procesa:

@Bean
public MessageChannel mqttInputChannel() {
    return new DirectChannel();
}

@Bean
public MessageProducer inbound(MqttPahoClientFactory factory, MqttProperties props) {
    MqttPahoMessageDrivenChannelAdapter adapter =
            new MqttPahoMessageDrivenChannelAdapter(props.clientId() + "-in", factory, props.inboundTopic());
    adapter.setCompletionTimeout(5000);
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(1);
    adapter.setOutputChannel(mqttInputChannel());
    return adapter;
}

@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler inboundHandler() {
    return message -> {
        String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class);
        System.out.println("received " + topic + ": " + message.getPayload());
    };
}

Headers útiles en cada mensaje entrante:

El converter por defecto convierte los payloads en String. Llama a converter.setPayloadAsBytes(true) si recibes formatos binarios como CBOR o Protobuf. También puedes añadir o quitar topics en tiempo de ejecución con adapter.addTopic(...) y adapter.removeTopic(...).

Errores y handlers lentos

Con un DirectChannel, tu handler se ejecuta en el hilo de callback de Paho. Eso tiene dos consecuencias que conviene planificar:

• Una excepción en el handler se envía al canal de errores del adapter si configuras uno con adapter.setErrorChannel(...). Sin él, la excepción se registra en el log y, según la versión y la configuración del adapter, Paho puede tratar el fallo como una conexión perdida. Captura y enruta siempre los errores de forma deliberada, por ejemplo hacia un canal que escriba los payloads fallidos en un topic de dead letter.
• Un handler lento retrasa el acknowledgement de los mensajes QoS 1 y bloquea todos los demás mensajes de esa conexión. Si el procesamiento incluye una base de datos o una llamada HTTP, cambia mqttInputChannel por un ExecutorChannel respaldado por un pool de hilos acotado, para que el hilo MQTT solo entregue el mensaje.

Ten en cuenta que pasar el mensaje a otro hilo significa que el PUBACK se envía antes de que termine tu procesamiento. Si el servicio se cae entre medias, ese mensaje se pierde. Cuando el procesamiento at-least-once sea realmente importante, usa adapter.setManualAcks(true) y confirma desde tu handler con el header IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK una vez que el trabajo esté confirmado.

Salida: publicar a través de un gateway

MqttPahoMessageHandler publica todo lo que llega a su canal de entrada. Combínalo con un @MessagingGateway para que el resto de tu código llame a una interfaz Java normal:

@Bean
public MessageChannel mqttOutboundChannel() {
    return new DirectChannel();
}

@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound(MqttPahoClientFactory factory, MqttProperties props) {
    MqttPahoMessageHandler handler = new MqttPahoMessageHandler(props.clientId() + "-out", factory);
    handler.setAsync(true);
    handler.setAsyncEvents(true);
    handler.setDefaultTopic(props.outboundTopic());
    handler.setDefaultQos(1);
    return handler;
}

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {

    void send(String payload);

    void send(String payload, @Header(MqttHeaders.TOPIC) String topic);

    void send(String payload, @Header(MqttHeaders.TOPIC) String topic,
              @Header(MqttHeaders.QOS) int qos, @Header(MqttHeaders.RETAINED) boolean retained);
}

Ahora cualquier bean puede inyectar MqttGateway y llamar a gateway.send("{\"id\":42}", "testmqtt/spring/out/orders"). setAsync(true) hace que el envío retorne sin esperar al PUBACK; con setAsyncEvents(true), Spring publica un MqttMessageDeliveredEvent cuando el broker confirma la entrega.

La alternativa con Java DSL

Si prefieres flujos a beans separados, el Java DSL expresa el mismo pipeline de entrada en pocas líneas, incluido un paso de transformación:

@Bean
public IntegrationFlow temperatureFlow(MqttPahoClientFactory factory, MqttProperties props) {
    return IntegrationFlow
            .from(new MqttPahoMessageDrivenChannelAdapter(
                    props.clientId() + "-temp", factory, "testmqtt/spring/sensors/+/temp"))
            .<String, Double>transform(Double::valueOf)
            .filter((Double t) -> t > 30.0)
            .handle(m -> System.out.println("too hot: " + m.getPayload()))
            .get();
}

Comprueba qué topics coinciden con un filtro como testmqtt/spring/sensors/+/temp en el topic matcher antes de desplegarlo.

Eventos de conexión y MQTT 5

Spring Integration publica eventos de aplicación para el ciclo de vida MQTT. Escucharlos es la forma más sencilla de registrar o alertar sobre problemas de conexión:

@Component
public class MqttEvents {

    @EventListener
    public void onFailure(MqttConnectionFailedEvent event) {
        System.err.println("MQTT connection failed: " + event.getCause());
    }

    @EventListener
    public void onSubscribed(MqttSubscribedEvent event) {
        System.out.println(event.getMessage());
    }
}

Para MQTT 5, cambia las clases: Mqttv5PahoMessageDrivenChannelAdapter y Mqttv5PahoMessageHandler reciben unas MqttConnectionOptions de Paho v5 en lugar de una factory, y mapean las user properties de MQTT 5 desde y hacia headers del mensaje. Los canales, el gateway y los handlers siguen igual.

Escalar con shared subscriptions

Con una suscripción normal, cada instancia de tu servicio recibe todos los mensajes: tres pods significan que cada pedido se procesa tres veces. Las shared subscriptions de MQTT 5 (que muchos brokers también admiten como extensión en conexiones MQTT 3.1.1) lo resuelven: suscríbete a $share/orders/testmqtt/spring/in/# y el broker reparte los mensajes entre todos los miembros del grupo orders. Solo cambia el topic en application.yml. Lee shared subscriptions de MQTT para conocer las reglas de entrega.

Pruébalo de extremo a extremo

1. Arranca la aplicación y busca la suscripción en el log.
2. Abre el cliente MQTT online, conéctate al mismo broker y publica hello en testmqtt/spring/in/demo. Tu servicio debería registrar received testmqtt/spring/in/demo: hello.
3. Suscríbete en el navegador a testmqtt/spring/out/#, llama a un endpoint que use MqttGateway y comprueba que el mensaje aparece con QoS 1.
4. Corta la conexión con el broker (por ejemplo, desactivando la red un momento) y confirma que se registra un MqttConnectionFailedEvent y que el adapter se reconecta.

Mira el tráfico de tu servicio Spring Boot

Si los mensajes no llegan, la excepción de Paho en el log suele explicar el motivo: errores de Paho Java recoge los más comunes, y la tabla de reason codes descifra los rechazos del broker.

Spring Integration convierte MQTT en un canal más de tu aplicación Spring Boot: configura la client factory una vez, conecta los adapters de entrada a service activators, publica a través de un gateway y mantén los client IDs únicos por conexión. Cuando la prueba desde el navegador pase, la misma configuración funcionará en producción cambiando solo las credenciales.

Preguntas frecuentes