MQTT (Message Queuing Telemetry Transport) es un protocolo de mensajería ligero diseñado para dispositivos IoT, basado en el modelo publicación/suscripción. Permite la comunicación eficiente entre aplicaciones de software y hardware en entornos de red limitados. Este artículo explica cómo implementar MQTT en Spring Boot para gestionar la recepción y envío de mensajes en proyectos de IoT.

Dependencia y Configuración

Para integrar MQTT en un proyecto Spring Boot, incluye la dependencia adecuada en el archivo pom.xml:

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

Configura las propiedades del servidor MQTT en application.yml, especificando la URL, credenciales y parámetros del cliente:

spring:
  mqtt:
    servidor: tcp://broker.emqx.io:1883
    credenciales:
      usuario: admin
      clave: 123456
    cliente:
      identificador: subscripcion-app
    tema:
      principal: datos/sensores

Este ejemplo utiliza el broker público de EMQX para pruebas iniciales. En producción, se recomienda configurar un servidor MQTT privado.

Implementación del Suscriptor

El suscriptor escucha mensajes de temas específicos. Primero, define una clase de configuración para establecer la conexión y suscribirse a temas:

@Configuration
public class ConfigSuscriptor {
    @Value("${spring.mqtt.servidor}")
    private String urlBroker;

    @Value("${spring.mqtt.credenciales.usuario}")
    private String usuario;

    @Value("${spring.mqtt.credenciales.clave}")
    private String clave;

    @Value("${spring.mqtt.cliente.identificador}")
    private String idCliente;

    @Value("${spring.mqtt.tema.principal}")
    private String temaPrincipal;

    private MqttClient clienteMqtt;

    @PostConstruct
    public void inicializar() {
        try {
            clienteMqtt = new MqttClient(urlBroker, idCliente, new MemoryPersistence());
            MqttConnectOptions opciones = new MqttConnectOptions();
            opciones.setCleanSession(true);
            opciones.setUserName(usuario);
            opciones.setPassword(clave.toCharArray());
            opciones.setConnectionTimeout(30);
            opciones.setKeepAliveInterval(60);
            clienteMqtt.setCallback(new ManejadorMensajes());
            clienteMqtt.connect(opciones);
            int[] qosNiveles = {1};
            clienteMqtt.subscribe(temaPrincipal, qosNiveles);
        } catch (MqttException e) {
            // Manejo de errores de conexión
        }
    }

    public void suscribirATema(String tema, int qos) {
        try {
            clienteMqtt.subscribe(tema, qos);
        } catch (MqttException e) {
            // Manejo de excepciones
        }
    }
}

Implementa un manejador de callbacks para procesar mensajes entrantes:

@Component
public class ManejadorMensajes implements MqttCallback {
    @Override
    public void connectionLost(Throwable causa) {
        // Lógica para reconexión automática
    }

    @Override
    public void messageArrived(String tema, MqttMessage mensaje) throws Exception {
        String contenido = new String(mensaje.getPayload());
        System.out.println("Tema: " + tema + ", Contenido: " + contenido);
        // Procesamiento adicional según el tema o contenido
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // Confirmación de entrega para mensajes publicados
    }
}

Implementación del Publicador

El publicador envía mensajes a temas específicos. Configura una clase separada para gestionar la conexión y publicar mensajes:

@Configuration
public class ConfigPublicador {
    @Value("${spring.mqtt.servidor}")
    private String urlBroker;

    @Value("${spring.mqtt.credenciales.usuario}")
    private String usuario;

    @Value("${spring.mqtt.credenciales.clave}")
    private String clave;

    private MqttClient clientePublicador;
    private String idPublicador = "publicador-app";

    @PostConstruct
    public void inicializar() {
        try {
            clientePublicador = new MqttClient(urlBroker, idPublicador, new MemoryPersistence());
            MqttConnectOptions opciones = new MqttConnectOptions();
            opciones.setUserName(usuario);
            opciones.setPassword(clave.toCharArray());
            clientePublicador.connect(opciones);
        } catch (MqttException e) {
            // Manejo de errores
        }
    }

    public void enviarMensaje(String tema, String contenido, int qos, boolean retener) {
        try {
            MqttMessage mensaje = new MqttMessage(contenido.getBytes());
            mensaje.setQos(qos);
            mensaje.setRetained(retener);
            MqttTopic temaDestino = clientePublicador.getTopic(tema);
            MqttDeliveryToken token = temaDestino.publish(mensaje);
            token.waitForCompletion();
        } catch (MqttException e) {
            // Manejo de excepciones en el envío
        }
    }
}

Parámetros Clave en MQTT

Topic (Tema): Es una cadena jerárquica que actúa como dirección para el enrutamiento de mensajes. Los niveles se separan por barras (por ejemplo, "dispositivos/actuadores/luces"). Los comodines "+" y "#" permiten suscripciones flexibles.

QoS (Calidad de Servicio): Define la garantía de entrega de mensajes. Los niveles son 0 (máximo una vez, sin confirmación), 1 (al menos una vez, con confirmación posible duplicado) y 2 (exactamente una vez, mediante protocolo de cuatro vías).

Retained (Retención): Cuando un mensaje se publica con la bandera de retención activada, el broker guarda el último mensaje para ese tema. Los nuevos suscriptores reciben inmediatamente este mensaje al conectarse.