Integración MQTT con Spring Boot: consume datos de dispositivos y exponlos por REST

Un servicio Spring Boot MQTT es una forma habitual de conectar una flota de dispositivos con el resto de los sistemas de tu negocio: los dispositivos publican por MQTT, y tus aplicaciones web y tus socios usan una API REST conocida. En este tutorial usarás los adaptadores MQTT 5 de Spring Integration para consumir telemetría por TLS, guardar la última lectura de cada dispositivo, exponerla en /api/devices y enviar comandos de vuelta a los dispositivos, todo a través de una única conexión MQTT compartida.

Arquitectura

El servicio se sitúa entre dos mundos:

El detalle más importante para producción es la conexión. Si el adaptador de entrada y el handler de salida crean cada uno su propio cliente MQTT, necesitan dos client ID y dos conexiones. Muchos brokers MQTT gestionados, ConnectMQ incluido, emiten un client ID vinculado a un conjunto de credenciales, por lo que dos conexiones con el mismo ID se desconectarían mutuamente sin parar. Spring Integration 6 resuelve esto con un ClientManager que comparten ambos componentes. Este tutorial lo usa desde el principio.

Configuración del proyecto

Crea un proyecto Spring Boot 3 (Java 17+) con estas dependencias en pom.xml:

<parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>3.3.5</version>
</parent>

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <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>
  <dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.mqttv5.client</artifactId>
    <version>1.2.5</version>
  </dependency>
</dependencies>

El cliente Paho MQTT 5 es una dependencia opcional de spring-integration-mqtt, así que debes añadirlo explícitamente. Olvidarlo es la causa más habitual de ClassNotFoundException: org.eclipse.paho.mqttv5... al arrancar.

Configuración

src/main/resources/application.yml:

mqtt:
  url: ${MQTT_URL:ssl://broker.connectmq.com:8883}
  client-id: ${MQTT_CLIENT_ID:spring-s8d2lf}
  username: ${MQTT_USERNAME:}
  password: ${MQTT_PASSWORD:}
  prefix: ${MQTT_PREFIX:connectmq/demo/s8d2lf}

Los valores por defecto apuntan al servidor MQTT público gratuito en broker.connectmq.com, que acepta clientes anónimos con cualquier client ID único y no vacío. Los topics del área pública son visibles para todo el mundo, así que usa un prefijo aleatorio y solo datos de prueba. El esquema ssl:// con el puerto 8883 activa TLS; el JDK confía de serie en el certificado de Let's Encrypt del broker.

La clase de configuración MQTT

package com.example.mqttdemo;

import java.nio.charset.StandardCharsets;

import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.mqtt.core.Mqttv5ClientManager;
import org.springframework.integration.mqtt.inbound.Mqttv5PahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.Mqttv5PahoMessageHandler;
import org.springframework.messaging.MessageHandler;

@Configuration
@IntegrationComponentScan
public class MqttConfig {

    @Bean
    public Mqttv5ClientManager mqttClientManager(
            @Value("${mqtt.url}") String url,
            @Value("${mqtt.client-id}") String clientId,
            @Value("${mqtt.username}") String username,
            @Value("${mqtt.password}") String password) {
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setServerURIs(new String[] { url });
        options.setAutomaticReconnect(true);
        options.setCleanStart(false);
        options.setSessionExpiryInterval(600L);
        options.setKeepAliveInterval(30);
        if (!username.isBlank()) {
            options.setUserName(username);
            options.setPassword(password.getBytes(StandardCharsets.UTF_8));
        }
        return new Mqttv5ClientManager(options, clientId);
    }

    @Bean
    public Mqttv5PahoMessageDrivenChannelAdapter mqttInbound(
            Mqttv5ClientManager clientManager,
            @Value("${mqtt.prefix}") String prefix) {
        Mqttv5PahoMessageDrivenChannelAdapter adapter =
                new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, prefix + "/devices/+/telemetry");
        adapter.setQos(1);
        return adapter;
    }

    @Bean
    public IntegrationFlow telemetryFlow(Mqttv5PahoMessageDrivenChannelAdapter mqttInbound,
                                         ReadingStore store) {
        return IntegrationFlow.from(mqttInbound)
                .handle(message -> store.accept(message))
                .get();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutbound")
    public MessageHandler mqttOutboundHandler(Mqttv5ClientManager clientManager) {
        Mqttv5PahoMessageHandler handler = new Mqttv5PahoMessageHandler(clientManager);
        handler.setAsync(true);
        handler.setDefaultQos(1);
        return handler;
    }
}

Tanto el adaptador como el handler reciben el mismo Mqttv5ClientManager, de modo que la aplicación abre exactamente una conexión MQTT con un único client ID. La reconexión automática está en las opciones de conexión; con setCleanStart(false) y un intervalo de expiración de sesión, el broker conserva la suscripción durante las caídas breves de red.

El gateway de salida

Un messaging gateway ofrece al resto de la aplicación una interfaz Java sencilla para publicar:

package com.example.mqttdemo;

import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;

@MessagingGateway(defaultRequestChannel = "mqttOutbound")
public interface MqttGateway {
    void publish(@Header(MqttHeaders.TOPIC) String topic, String payload);
}

El handler v5 convierte un payload String en bytes UTF-8, y la cabecera MqttHeaders.TOPIC fija el topic de destino de cada mensaje.

Almacenamiento de las últimas lecturas

package com.example.mqttdemo;

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Component
public class ReadingStore {

    public record DeviceReading(String deviceId, JsonNode data, Instant receivedAt) { }

    private static final Logger log = LoggerFactory.getLogger(ReadingStore.class);
    private final Map<String, DeviceReading> latest = new ConcurrentHashMap<>();
    private final ObjectMapper mapper;

    public ReadingStore(ObjectMapper mapper) {
        this.mapper = mapper;
    }

    public void accept(Message<?> message) {
        String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class);
        String[] parts = topic.split("/");
        String deviceId = parts[parts.length - 2];
        String json = new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);
        try {
            latest.put(deviceId, new DeviceReading(deviceId, mapper.readTree(json), Instant.now()));
        } catch (JsonProcessingException e) {
            log.warn("Ignoring invalid JSON from {}: {}", deviceId, e.getOriginalMessage());
        }
    }

    public Collection<DeviceReading> all() {
        return latest.values();
    }

    public Optional<DeviceReading> get(String deviceId) {
        return Optional.ofNullable(latest.get(deviceId));
    }
}

El adaptador de entrada entrega payloads byte[] por defecto, y el topic recibido llega en la cabecera MqttHeaders.RECEIVED_TOPIC. El ID del dispositivo es el segmento justo anterior a telemetry.

El controlador REST

package com.example.mqttdemo;

import java.util.Collection;
import java.util.regex.Pattern;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/devices")
public class DeviceController {

    private static final Pattern SAFE_ID = Pattern.compile("[A-Za-z0-9_-]{1,64}");

    private final ReadingStore store;
    private final MqttGateway gateway;
    private final String prefix;

    public DeviceController(ReadingStore store, MqttGateway gateway,
                            @Value("${mqtt.prefix}") String prefix) {
        this.store = store;
        this.gateway = gateway;
        this.prefix = prefix;
    }

    @GetMapping
    public Collection<ReadingStore.DeviceReading> all() {
        return store.all();
    }

    @GetMapping("/{id}")
    public ResponseEntity<ReadingStore.DeviceReading> one(@PathVariable String id) {
        return ResponseEntity.of(store.get(id));
    }

    @PostMapping("/{id}/commands")
    public ResponseEntity<Void> command(@PathVariable String id, @RequestBody String body) {
        if (!SAFE_ID.matcher(id).matches()) {
            return ResponseEntity.badRequest().build();
        }
        gateway.publish(prefix + "/devices/" + id + "/cmd", body);
        return ResponseEntity.accepted().build();
    }
}

La comprobación del ID no es decorativa. Sin ella, quien llame a la API podría enviar un ID que contenga /, + o # y publicar en topics que nunca quisiste exponer.

Cómo fluye un mensaje por la aplicación

Ayuda seguir un mensaje de telemetría de principio a fin, porque Spring Integration oculta mucha fontanería detrás de unos pocos beans.

1. El dispositivo publica {"temperature":22.5} en <prefix>/devices/sensor-1/telemetry.
2. El broker lo enruta a todas las suscripciones que coinciden, incluida la que crea el adaptador de entrada a través del client manager compartido.
3. El adaptador envuelve los bytes del payload en un Message de Spring y añade cabeceras como RECEIVED_TOPIC, RECEIVED_QOS y RECEIVED_RETAINED.
4. El telemetryFlow pasa el mensaje a ReadingStore.accept, que analiza el JSON y sustituye la lectura anterior de ese dispositivo.
5. Una llamada REST a /api/devices/sensor-1 lee el mapa y devuelve la última lectura como JSON.

Los comandos viajan en sentido contrario: el controlador llama al gateway, el gateway envía un mensaje al canal mqttOutbound, el service activator lo pasa al handler de salida y el handler publica a través de la misma conexión compartida. Como el handler de salida es asíncrono, la llamada REST devuelve 202 Accepted sin esperar la confirmación del broker. Si tu API debe confirmar la entrega, cambia el handler a modo síncrono y asume la latencia adicional.

Gestión de errores

Por defecto, una excepción dentro del flujo se registra en el log y el mensaje se descarta. Para cualquier cosa que vaya más allá de un prototipo, añade un canal de errores al adaptador de entrada con adapter.setErrorChannelName("mqttErrors") y un pequeño flujo que registre el topic y el payload, para que un mensaje mal formado de un dispositivo sea visible en lugar de ignorarse en silencio. Mantén el handler rápido: si añades escrituras en base de datos, plantéate usar un executor channel entre el adaptador y el almacén para que una base de datos lenta no retrase la entrega de mensajes.

Prueba de extremo a extremo

Arranca la aplicación con ./mvnw spring-boot:run y simula un dispositivo:

mosquitto_pub -h broker.connectmq.com -p 1883 -i dev-s8d2lf \
  -t 'connectmq/demo/s8d2lf/devices/sensor-1/telemetry' -m '{"temperature":22.5,"humidity":48}'

Consulta la API REST:

curl -s http://localhost:8080/api/devices
curl -s http://localhost:8080/api/devices/sensor-1

Envía un comando y observa cómo llega al topic del dispositivo:

mosquitto_sub -h broker.connectmq.com -p 1883 -i cmdwatch-s8d2lf \
  -t 'connectmq/demo/s8d2lf/devices/+/cmd' -v &
curl -s -X POST http://localhost:8080/api/devices/sensor-1/commands \
  -H 'Content-Type: application/json' -d '{"fan":"on"}'

Notas de producción para Spring Boot MQTT

1. Usa las credenciales emitidas. Crea un broker en el portal de ConnectMQ y asigna a MQTT_CLIENT_ID, MQTT_USERNAME y MQTT_PASSWORD los valores generados; el client ID debe coincidir exactamente.
2. Una instancia, un client ID. Si escalas a varias réplicas, cada una necesita sus propias credenciales. Para repartir el consumo de la misma telemetría, las suscripciones compartidas de MQTT 5 están disponibles en los planes Serverless de pago.
3. Persiste lo importante. El mapa en memoria se pierde al reiniciar. Para tener histórico, escribe las lecturas en una base de datos dentro del flujo o suscribe una herramienta de ingesta dedicada junto a este servicio.
4. Elige el QoS a conciencia. QoS 1 es el valor por defecto adecuado para telemetría y comandos; consulta los niveles de QoS de MQTT explicados. El plan Free admite QoS hasta 1.
5. Expón el estado de salud. Añade Spring Boot Actuator y genera una alerta cuando la conexión MQTT esté caída más tiempo del esperado.

Solución de problemas

La referencia completa de los adaptadores está en la documentación MQTT de Spring Integration.

Guías relacionadas

Si necesitas un cliente Java sin Spring, el tutorial de cliente MQTT en Java compara Paho con el cliente de HiveMQ. El tutorial de cliente MQTT en Go construye el mismo patrón en Go, y la guía de bibliotecas cliente MQTT cubre otros ecosistemas.

Conecta Spring Boot a ConnectMQ

Un broker MQTT gestionado da a tu servicio Spring Boot un endpoint estable y cifrado mientras tú te centras en la API. ConnectMQ admite MQTT 5 con expiración de sesión y suscripciones compartidas en los planes de pago, y su vista de dispositivos y su explorador de topics en vivo te ayudan a comprobar qué está enviando tu flota. Para un repaso más amplio de las opciones de alojamiento, consulta la guía de brokers MQTT en la nube. Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y configura las variables de entorno anteriores.

Preguntas frecuentes

¿Cómo uso MQTT en Spring Boot?

Añade spring-integration-mqtt y el cliente Paho MQTT 5, y configura un Mqttv5PahoMessageDrivenChannelAdapter para los mensajes entrantes y un Mqttv5PahoMessageHandler para los salientes.

¿Pueden los adaptadores MQTT de entrada y salida compartir una conexión?

Sí. Desde Spring Integration 6, pasa el mismo Mqttv5ClientManager a ambos, que usa un único client ID y una única conexión.

¿Spring Integration admite MQTT 5?

Sí, mediante los adaptadores con prefijo Mqttv5, que se basan en el cliente Eclipse Paho MQTT 5.

¿Cómo conecto Spring Boot a un broker MQTT por TLS?

Usa una URI de servidor ssl://host:8883 en las opciones de conexión. Los brokers con certificados de confianza pública funcionan con el trust store por defecto del JDK.