Introducción a MQTT
MQTT (Message Queuing Telemetry Transport) es un protocolo de mensajería ligero basado en el modelo Publicar/Suscribir. Es ideal para la comunicación en Internet de las Cosas (IoT) debido a su simplicidad, eficiencia y bajo consumo de recursos. Admite niveles de Calidad de Servicio (QoS) y maneja mensajes de tamaño reducido, lo que lo hace adecuado para dispositivos con capacidades de cómputo limitadas y redes con bajo ancho de banda.
Características Clave de MQTT
- Modelo Publicar/Suscribir: Permite la distribución de mensajes uno a muchos, desacoplando los componantes de la aplicación.
- Abstracción de Carga Útil: El broker no interpreta el contenido de los mensajes.
- Transporte sobre TCP/IP: Utiliza TCP para una conexión de red fiable.
- Calidad de Servicio (QoS):
- Como mucho una vez (At most once): El mensaje se envía, pero no hay confirmación. Es posible la pérdida o duplicación. Adecuado para datos de sensores donde la pérdida ocasional no es crítica.
- Al menos una vez (At least once): Se garantiza que el mensaje llegue, pero puede haber duplicados.
- Exactamente una vez (Exactly once): Se garantiza que el mensaje llegue una sola vez. Fundamental para aplicaciones críticas como sistemas de facturación.
- Mensajse de Tamaño Reducido: Con una cabecera de tamaño fijo de 2 bytes, minimiza el tráfico de red.
- Last Will and Testament: Un mecanismo para notificar a otras partes sobre la desconexión inesperada de un cliente.
Escenarios de Aplicación de MQTT
Gracias a su bajo consumo de ancho de banda y recursos, MQTT es ampliamente utilizado en:
- Notificaciones Push: Para aplicaciones móviles y de escritorio.
- Pedidos Inteligentes: Facilitando la toma de pedidos en restaurantes a través de códigos QR.
- Actualización de Información: Gestionando la visualización en etiquetas electrónicas o pantallas multimedia en tiendas o espacios públicos.
- Control de Acceso y Pagos: Como en sistemas de peaje de estacionamientos o torniquetes de metro que utilizan códigos QR.
Componentes de MQTT
1. Servidor (Broker)
El Broker es el intermediario central que recibe todos los mensajes de los publicadores y los reenvía a los suscriptores correspondientes. EMQX es un ejemplo popular de broker MQTT, desarrollado en Erlang. Otros brokers incluyen ActiveMQ y RabbitMQ.
- EMQX Broker: https://www.emqx.io/zh/downloads?os=Windows
2. Cliente (Publicador/Suscriptor)
Los clientes son las aplicaciones que envían (publican) o reciben (se suscriben) mensajes a través del broker. MQTTX es una herrramienta útil para probar clientes MQTT.
- MQTTX Client: https://mqttx.app/zh
Integración con Spring Boot
Añadir Dependencia de MQTT
Para integrar MQTT en una aplicación Spring Boot, se puede utilizar la dependencia spring-integration-mqtt.
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
Implementación del Publicador (Productor)
Configuración de la Aplicación (application.properties o application.yml)
Se deben definir los parámetros de conexión al broker MQTT.
server:
port: 8081
spring:
application:
name: provider
mqtt:
url: tcp://127.0.0.1:11883 # Dirección del broker MQTT
username: admin
password: 123456
client:
id: provider-client-id # ID único del cliente publicador
default:
topic: iot/data # Tema predeterminado para publicar mensajes
Configuración del Cliente Publicador
Esta clase configura el cliente MQTT, establece la conexión y define el método para publicar mensajes.
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import lombok.extern.slf4j.Slf4j;
import jakarta.annotation.PostConstruct;
@Configuration
@Slf4j
public class MqttPublisherConfig {
@Value("${spring.mqtt.username}")
private String mqttUsername;
@Value("${spring.mqtt.password}")
private String mqttPassword;
@Value("${spring.mqtt.url}")
private String mqttBrokerUrl;
@Value("${spring.mqtt.client.id}")
private String mqttClientId;
private MqttClient mqttClient;
@PostConstruct
public void initializeConnection() {
connectToBroker();
}
public void connectToBroker() {
try {
mqttClient = new MqttClient(mqttBrokerUrl, mqttClientId, new MemoryPersistence());
MqttConnectOptions connectOptions = new MqttConnectOptions();
connectOptions.setCleanSession(true); // Establecer session limpia
connectOptions.setUserName(mqttUsername);
connectOptions.setPassword(mqttPassword.toCharArray());
connectOptions.setConnectionTimeout(100); // Tiempo de espera de conexión en segundos
connectOptions.setKeepAliveInterval(20); // Intervalo de keep-alive en segundos
// Configuración del mensaje "Last Will"
String willTopic = "will/topic";
String willMessage = mqttClientId + " disconnected unexpectedly.";
connectOptions.setWill(willTopic, willMessage.getBytes(), 0, false);
mqttClient.setCallback(new MqttPublisherCallback());
if (!mqttClient.isConnected()) {
mqttClient.connect(connectOptions);
log.info("Publisher client '{}' connected to MQTT broker.", mqttClientId);
}
} catch (MqttException e) {
log.error("Failed to connect publisher client '{}' to MQTT broker: {}", mqttClientId, e.getMessage());
e.printStackTrace();
}
}
public void publishMessage(int qos, boolean retained, String topic, String payload) {
if (mqttClient == null || !mqttClient.isConnected()) {
log.warn("Publisher client '{}' is not connected. Cannot publish message.", mqttClientId);
// Opcionalmente, intentar reconectar aquí
return;
}
try {
MqttMessage message = new MqttMessage();
message.setQos(qos);
message.setRetained(retained);
message.setPayload(payload.getBytes());
MqttTopic mqttTopic = mqttClient.getTopic(topic);
MqttDeliveryToken token = mqttTopic.publish(message);
token.waitForCompletion(); // Esperar a que la entrega se complete
log.debug("Message published to topic '{}' by client '{}'.", topic, mqttClientId);
} catch (MqttException e) {
log.error("Failed to publish message to topic '{}' by client '{}': {}", topic, mqttClientId, e.getMessage());
e.printStackTrace();
}
}
}
Callback del Cliente Publicador
Maneja eventos como la pérdida de conexión y la entrega de mensajes.
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import lombok.extern.slf4j.Slf4j;
@Configuration
@Slf4j
public class MqttPublisherCallback implements MqttCallback {
@Value("${spring.mqtt.client.id}")
private String clientId;
@Override
public void connectionLost(Throwable cause) {
log.warn("Publisher client '{}' connection lost: {}", clientId, cause.getMessage());
// Implementar lógica de reconexión si es necesario
}
@Override
public void messageArrived(String topic, org.eclipse.paho.client.mqttv3.MqttMessage message) {
// No se espera que el publicador reciba mensajes en este contexto
log.debug("Publisher client '{}' received unexpected message on topic '{}': {}", clientId, topic, new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
// El mensaje fue publicado exitosamente
log.info("Publisher client '{}' message delivery confirmed.", clientId);
}
}
Controlador para Probar el Publicador
Un endpoint HTTP para desencadenar la publicación de mensajes.
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
@Controller
public class PublisherTestController {
@Autowired
private MqttPublisherConfig publisherClient;
@RequestMapping("/publish")
@ResponseBody
public String sendMessage(
@RequestParam(defaultValue = "1") int qos,
@RequestParam(defaultValue = "false") boolean retained,
@RequestParam String topic,
@RequestParam String message) {
try {
publisherClient.publishMessage(qos, retained, topic, message);
return "Message published successfully to topic: " + topic;
} catch (Exception e) {
e.printStackTrace();
return "Failed to publish message: " + e.getMessage();
}
}
}
Implementación del Suscriptor (Consumidor)
Configuración de la Aplicación (application.properties o application.yml)
Similar al publicador, pero con un ID de cliente distinto.
server:
port: 8082
spring:
application:
name: consumer
mqtt:
url: tcp://127.0.0.1:11883 # Dirección del broker MQTT
username: admin
password: 123456
client:
id: consumer-client-id # ID único del cliente suscriptor
default:
topic: iot/# # Tema predeterminado para suscribirse (ej. suscripción a todos los subtemas)
Configuración del Cliente Suscriptor
Configura el cliente MQTT, se conecta al broker y se suscribe a los temas deseados.
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import lombok.extern.slf4j.Slf4j;
import jakarta.annotation.PostConstruct;
@Configuration
@Slf4j
public class MqttSubscriberConfig {
@Value("${spring.mqtt.username}")
private String mqttUsername;
@Value("${spring.mqtt.password}")
private String mqttPassword;
@Value("${spring.mqtt.url}")
private String mqttBrokerUrl;
@Value("${spring.mqtt.client.id}")
private String mqttClientId;
@Value("${spring.mqtt.default.topic}")
private String defaultSubscriptionTopic;
private MqttClient mqttClient;
@PostConstruct
public void initializeConnection() {
connectToBrokerAndSubscribe();
}
public void connectToBrokerAndSubscribe() {
try {
mqttClient = new MqttClient(mqttBrokerUrl, mqttClientId, new MemoryPersistence());
MqttConnectOptions connectOptions = new MqttConnectOptions();
connectOptions.setCleanSession(true); // Establecer session limpia
connectOptions.setUserName(mqttUsername);
connectOptions.setPassword(mqttPassword.toCharArray());
connectOptions.setConnectionTimeout(100);
connectOptions.setKeepAliveInterval(20);
connectOptions.setWill("will/topic", (mqttClientId + " disconnected unexpectedly.").getBytes(), 0, false);
mqttClient.setCallback(new MqttSubscriberCallback());
if (!mqttClient.isConnected()) {
mqttClient.connect(connectOptions);
log.info("Subscriber client '{}' connected to MQTT broker.", mqttClientId);
subscribeToTopics();
}
} catch (MqttException e) {
log.error("Failed to connect subscriber client '{}' to MQTT broker: {}", mqttClientId, e.getMessage());
e.printStackTrace();
}
}
private void subscribeToTopics() {
try {
// Ejemplo de suscripción a múltiples temas con diferentes QoS
String[] topics = {defaultSubscriptionTopic, "sensor/+/temperature"};
int[] qos = {1, 1}; // QoS para cada tema
mqttClient.subscribe(topics, qos);
log.info("Subscriber client '{}' subscribed to topics: {}", mqttClientId, String.join(", ", topics));
} catch (MqttException e) {
log.error("Failed to subscribe to topics for client '{}': {}", mqttClientId, e.getMessage());
e.printStackTrace();
}
}
public void disconnectFromBroker() {
try {
if (mqttClient != null && mqttClient.isConnected()) {
mqttClient.disconnect();
log.info("Subscriber client '{}' disconnected from MQTT broker.", mqttClientId);
}
} catch (MqttException e) {
log.error("Failed to disconnect subscriber client '{}': {}", mqttClientId, e.getMessage());
e.printStackTrace();
}
}
// Método para suscripción manual
public void subscribe(String topic, int qos) {
if (mqttClient == null || !mqttClient.isConnected()) {
log.warn("Subscriber client '{}' is not connected. Cannot subscribe to topic '{}'.", mqttClientId, topic);
return;
}
try {
mqttClient.subscribe(topic, qos);
log.info("Subscriber client '{}' subscribed to topic '{}' with QoS {}.", mqttClientId, topic, qos);
} catch (MqttException e) {
log.error("Failed to subscribe to topic '{}' for client '{}': {}", topic, mqttClientId, e.getMessage());
e.printStackTrace();
}
}
}
Callback del Cliente Suscriptor
Maneja la recepción de mensajes y la pérdida de conexión.
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.context.annotation.Configuration;
import lombok.extern.slf4j.Slf4j;
@Configuration
@Slf4j
public class MqttSubscriberCallback implements MqttCallback {
@Override
public void connectionLost(Throwable cause) {
log.warn("Subscriber client connection lost: {}", cause.getMessage());
// Implementar lógica de reconexión
}
@Override
public void messageArrived(String topic, MqttMessage message) {
log.info("Message arrived:");
log.info(" Topic: {}", topic);
log.info(" QoS: {}", message.getQos());
log.info(" Retained: {}", message.isRetained());
log.info(" Payload: {}", new String(message.getPayload()));
// Aquí se procesaría el mensaje recibido (ej. guardar en base de datos, procesar lógica de negocio)
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
// El suscriptor no publica, por lo que este callback no suele ser relevante para él.
log.debug("Delivery complete for subscriber (should not happen unless publishing).");
}
}
Controlador para Gestionar Conexión/Desconexión
Endpoints para controlar manualmente la conexión y desconexión del suscriptor.
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
@Controller
public class SubscriberTestController {
@Autowired
private MqttSubscriberConfig subscriberClient;
@Value("${spring.mqtt.client.id}")
private String clientId;
@RequestMapping("/connectSubscriber")
@ResponseBody
public String connectClient() {
// La conexión ya se establece en @PostConstruct, este método podría ser para reconexión.
// Para este ejemplo, simplemente confirmamos que está configurado.
// Si se necesitara reconexión explícita: subscriberClient.connectToBrokerAndSubscribe();
return clientId + " is configured to connect/reconnect to the broker.";
}
@RequestMapping("/disconnectSubscriber")
@ResponseBody
public String disconnectClient() {
subscriberClient.disconnectFromBroker();
return clientId + " disconnected from the broker.";
}
@RequestMapping("/subscribeManual")
@ResponseBody
public String subscribeManually(String topic, int qos) {
subscriberClient.subscribe(topic, qos);
return clientId + " subscribed manually to topic: " + topic + " with QoS: " + qos;
}
}