MQTT en Java con HiveMQ MQTT Client

HiveMQ MQTT Client es una librería Java open source para MQTT 3.1.1 y MQTT 5 con tres estilos de API: bloqueante, asíncrona (CompletableFuture) y reactiva (RxJava, con un módulo opcional para Reactor). A pesar del nombre, funciona con cualquier broker que cumpla el estándar. En este tutorial lo conectas por TLS, publicas y te suscribes con cada estilo de API, usas funciones de MQTT 5 y verificas todo desde fuera.

Añadir la dependencia

El cliente necesita Java 8 o superior. Con Gradle:

dependencies {
    implementation("com.hivemq:hivemq-mqtt-client:1.3.3")
    // Only if you connect over WebSockets:
    // implementation("com.hivemq:hivemq-mqtt-client-websocket:1.3.3")
}

Con Maven:

<dependency>
  <groupId>com.hivemq</groupId>
  <artifactId>hivemq-mqtt-client</artifactId>
  <version>1.3.3</version>
</dependency>

Consulta Maven Central para ver la versión 1.3.x más reciente. Funciones opcionales como el transporte WebSocket y la API de Reactor están en artefactos separados, así el jar principal sigue siendo pequeño.

Crear un cliente

Todo cliente empieza con MqttClient.builder(). El builder es inmutable y fluido: eliges la versión del protocolo, el identificador de cliente, el servidor y el transporte, y luego llamas a buildBlocking(), buildAsync() o buildRx().

import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
import java.util.UUID;

Mqtt5BlockingClient client = MqttClient.builder()
        .useMqttVersion5()
        .identifier("testmqtt-java-" + UUID.randomUUID().toString().substring(0, 8))
        .serverHost("broker.connectmq.com")
        .serverPort(8883)
        .sslWithDefaultConfig()                 // TLS con el trust store de la JVM
        .automaticReconnectWithDefaultConfig()  // backoff exponencial
        .buildBlocking();

Conviene saber algunas cosas desde el principio:

• Los identificadores de cliente deben ser únicos por conexión. Dos procesos con el mismo ID se desconectan mutuamente en bucle. Genera uno con el generador de client ID o derívalo de un nombre estable de dispositivo o instancia.
• sslWithDefaultConfig() usa el trust store por defecto de la JVM, que ya confía en las CAs públicas. Para una CA privada, pasa un TrustManagerFactory con .sslConfig().trustManagerFactory(tmf).applySslConfig().
• Para MQTT 3.1.1 usa useMqttVersion3(): el resto del código es casi igual, sin las opciones de MQTT 5.

Conectar con credenciales

La conexión no se abre hasta que envías un CONNECT. Con la API bloqueante, send() devuelve el CONNACK o lanza una excepción:

import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException;
import java.nio.charset.StandardCharsets;

try {
    client.connectWith()
            .simpleAuth()
                .username(System.getenv("MQTT_USER"))
                .password(System.getenv("MQTT_PASS").getBytes(StandardCharsets.UTF_8))
                .applySimpleAuth()
            .cleanStart(false)
            .sessionExpiryInterval(3600)   // conservar la sesión una hora sin conexión
            .keepAlive(30)
            .send();
    System.out.println("connected");
} catch (Mqtt5ConnAckException e) {
    System.err.println("rejected: " + e.getMqttMessage().getReasonCode());
}

Si el broker rechaza la conexión, la excepción incluye el reason code de MQTT 5, por ejemplo BAD_USER_NAME_OR_PASSWORD o NOT_AUTHORIZED. Busca el valor en la tabla de reason codes de MQTT o lee los reason codes de CONNACK explicados para entender cada uno.

API bloqueante: publicar y recibir

La API bloqueante es la más fácil de leer e ideal para herramientas de línea de comandos y tests. Para recibir mensajes abres un stream Publishes antes de suscribirte, así no se pierde nada:

import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

try (Mqtt5BlockingClient.Mqtt5Publishes publishes =
             client.publishes(MqttGlobalPublishFilter.ALL)) {

    client.subscribeWith()
            .topicFilter("testmqtt/java/demo/#")
            .qos(MqttQos.AT_LEAST_ONCE)
            .send();

    client.publishWith()
            .topic("testmqtt/java/demo/hello")
            .qos(MqttQos.AT_LEAST_ONCE)
            .payload("hello from Java".getBytes(StandardCharsets.UTF_8))
            .send();

    Optional<Mqtt5Publish> msg = publishes.receive(10, TimeUnit.SECONDS);
    msg.ifPresent(p -> System.out.println(
            p.getTopic() + " -> " + new String(p.getPayloadAsBytes(), StandardCharsets.UTF_8)));
}
client.disconnect();

Como el stream es un recurso, el bloque try-with-resources lo cierra limpiamente. En un test, un Optional vacío tras el timeout es una señal de fallo clara.

API asíncrona: cadenas de CompletableFuture

En servicios, la API asíncrona evita bloquear hilos. Cada operación devuelve un CompletableFuture, así que puedes encadenar conexión, suscripción y publicación:

import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;

Mqtt5AsyncClient async = client.toAsync();   // misma conexión, otra API

async.subscribeWith()
        .topicFilter("testmqtt/java/sensors/+/temp")
        .qos(MqttQos.AT_LEAST_ONCE)
        .callback(p -> System.out.println(
                p.getTopic() + " " + new String(p.getPayloadAsBytes(), StandardCharsets.UTF_8)))
        .send()
        .thenCompose(subAck -> async.publishWith()
                .topic("testmqtt/java/sensors/lab/temp")
                .qos(MqttQos.AT_LEAST_ONCE)
                .payload("21.7".getBytes(StandardCharsets.UTF_8))
                .send())
        .whenComplete((result, err) -> {
            if (err != null) err.printStackTrace();
            else System.out.println("published");
        });

El callback que pasas a subscribeWith() se ejecuta en el event loop de Netty del cliente. Mantenlo corto: delega el trabajo pesado (escrituras en base de datos, llamadas HTTP) a tu propio executor, o ralentizarás todos los demás mensajes de esa conexión. Para elegir filtros de topic con seguridad, pruébalos antes en el topic matcher.

API reactiva: streams con backpressure

Mqtt5RxClient expone las suscripciones como Flowable de RxJava. Es útil cuando el consumidor puede ser más lento que el productor, porque el backpressure se propaga al control de flujo de MQTT:

import com.hivemq.client.mqtt.mqtt5.Mqtt5RxClient;

Mqtt5RxClient rx = client.toRx();

rx.subscribePublishesWith()
        .topicFilter("testmqtt/java/events/#")
        .qos(MqttQos.AT_LEAST_ONCE)
        .applySubscribe()
        .doOnSingle(subAck -> System.out.println("subscribed: " + subAck.getReasonCodes()))
        .map(p -> new String(p.getPayloadAsBytes(), StandardCharsets.UTF_8))
        .subscribe(System.out::println, Throwable::printStackTrace);

Si tu stack usa Project Reactor (Spring WebFlux, por ejemplo), añade el artefacto hivemq-mqtt-client-reactor y trabaja con Flux y Mono.

Funciones de MQTT 5

Los builders exponen directamente las propiedades de MQTT 5, sin necesidad de manipular mapas de propiedades en bruto:

import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PayloadFormatIndicator;

async.publishWith()
        .topic("testmqtt/java/orders/created")
        .qos(MqttQos.AT_LEAST_ONCE)
        .payload("{\"id\":42}".getBytes(StandardCharsets.UTF_8))
        .contentType("application/json")
        .payloadFormatIndicator(Mqtt5PayloadFormatIndicator.UTF_8)
        .messageExpiryInterval(60)                    // descartar si no se entrega en 60 s
        .responseTopic("testmqtt/java/orders/replies")
        .correlationData("req-42".getBytes(StandardCharsets.UTF_8))
        .userProperties()
            .add("source", "billing-service")
            .add("schema", "v2")
            .applyUserProperties()
        .send();

Más información en user properties de MQTT y probar funciones de MQTT 5.

Reconexiones y listeners de ciclo de vida

automaticReconnectWithDefaultConfig() reintenta con backoff exponencial. Para ajustarlo y registrar los cambios de estado, configúralo explícitamente:

Mqtt5AsyncClient resilient = MqttClient.builder()
        .useMqttVersion5()
        .identifier("testmqtt-java-worker-1")
        .serverHost("broker.connectmq.com")
        .serverPort(8883)
        .sslWithDefaultConfig()
        .automaticReconnect()
            .initialDelay(1, TimeUnit.SECONDS)
            .maxDelay(30, TimeUnit.SECONDS)
            .applyAutomaticReconnect()
        .addConnectedListener(ctx -> System.out.println("connected"))
        .addDisconnectedListener(ctx -> System.out.println("disconnected: " + ctx.getCause().getMessage()))
        .buildAsync();

Combina las reconexiones con cleanStart(false) y un intervalo de expiración de sesión: el broker conserva tus suscripciones y encola los mensajes QoS 1/2 mientras el cliente no está. Para eso es imprescindible un identificador de cliente estable: un ID aleatorio en cada arranque significa una sesión nueva y vacía cada vez. Ten en cuenta que el endpoint compartido de pruebas broker.connectmq.com no conserva sesiones offline, así que ejecuta las pruebas de persistencia de sesión contra tu propia instancia privada de broker.

WebSockets y apagado limpio

Algunas redes solo permiten tráfico HTTPS. Con el artefacto hivemq-mqtt-client-websocket en el classpath, el mismo cliente puede tunelizar MQTT por WebSockets seguros; solo cambia la configuración del transporte:

Mqtt5BlockingClient wsClient = MqttClient.builder()
        .useMqttVersion5()
        .identifier("testmqtt-java-ws")
        .serverHost("broker.connectmq.com")
        .serverPort(8084)
        .sslWithDefaultConfig()
        .webSocketConfig()
            .serverPath("mqtt")
            .applyWebSocketConfig()
        .buildBlocking();

Cuando tu aplicación se detenga, llama a disconnect() en lugar de simplemente salir de la JVM. Un DISCONNECT limpio indica al broker que no publique tu mensaje Will y, con MQTT 5, incluso puedes cambiar la expiración de sesión en ese momento con disconnectWith().sessionExpiryInterval(0).send() cuando la sesión ya no sea necesaria. En una aplicación Spring o Jakarta EE, engancha esto al callback de apagado del contenedor para que los despliegues progresivos no dejen sesiones fantasma ni disparen falsas alertas de "offline".

Pruébalo desde fuera

No confíes solo en tu propio cliente para saber que funciona. Abre el cliente MQTT online, conéctate al mismo broker y suscríbete a testmqtt/java/#. Ejecuta cada ejemplo y confirma que los mensajes llegan con el topic, el QoS y las propiedades de MQTT 5 esperados. Después publica desde el navegador en testmqtt/java/sensors/lab/temp y observa cómo tu callback asíncrono lo imprime.

Mira en directo los mensajes de tu cliente Java

Para comprobaciones automatizadas, la API bloqueante encaja muy bien con JUnit: consulta tests de integración MQTT con JUnit 5. Si todavía estás eligiendo librería, clientes MQTT para Java comparados pone este cliente junto a Eclipse Paho.

Errores comunes

• No se recibe nada en el ejemplo bloqueante: el stream publishes() se abrió después de la suscripción o el filtro global no coincide. Ábrelo primero.
• SSLHandshakeException: normalmente falta una CA privada en el trust store, o te conectas con TLS al puerto en claro 1883. Usa el 8883 para TLS.
• Los callbacks se atascan: código lento dentro del callback bloquea el event loop. Delégalo.
• Sesión perdida tras reiniciar: client ID aleatorio o cleanStart(true). Usa un ID estable y un intervalo de expiración de sesión.

HiveMQ MQTT Client le da a Java un cliente moderno con soporte completo de MQTT 5 en el estilo de API que prefieras. Empieza con la API bloqueante para validar la conexión, pasa a la asíncrona o la reactiva para el código de producción y mantén un cliente externo abierto mientras desarrollas para ver siempre lo que realmente pasa por el broker.

Preguntas frecuentes