Tutorial de cliente MQTT en Java: Eclipse Paho frente a HiveMQ Client con TLS y MQTT 5

Elegir un cliente MQTT en Java suele reducirse a dos bibliotecas: Eclipse Paho y HiveMQ MQTT Client. Este tutorial las compara con honestidad y después implementa el mismo programa de publicación/suscripción con cada una, usando MQTT 5 sobre TLS en el puerto 8883, reconexión automática y un Last Will, para que elijas la que mejor encaje en tu código.

Comparativa de bibliotecas cliente MQTT para Java

Ambas bibliotecas son de código abierto, ambas hablan MQTT 5 y ambas se usan en producción. Se diferencian sobre todo en el estilo de la API y en lo activo que es su desarrollo.

Una regla práctica: Paho es pequeño, no tiene dependencias y resulta familiar gracias a incontables ejemplos, lo que lo hace atractivo para Android o entornos con recursos ajustados. El cliente de HiveMQ tiene una API más moderna y gestiona bien la contrapresión con tasas de mensajes altas, lo que encaja con servicios de backend. Los desarrolladores de Spring suelen usar Paho de forma indirecta a través de Spring Integration; consulta la guía de MQTT con Spring Boot para esa vía.

Configuración del proyecto

Ambos ejemplos usan Java 17 y Maven. Añade una o ambas dependencias a pom.xml:

<dependencies>
  <dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.mqttv5.client</artifactId>
    <version>1.2.5</version>
  </dependency>
  <dependency>
    <groupId>com.hivemq</groupId>
    <artifactId>hivemq-mqtt-client</artifactId>
    <version>1.3.3</version>
  </dependency>
</dependencies>

Para las pruebas, conéctate al servidor MQTT público gratuito en broker.connectmq.com. Acepta clientes anónimos con cualquier client ID único y no vacío. Usa un prefijo aleatorio como connectmq/demo/j6t1ve y ten en cuenta que cualquiera puede leer los topics públicos: es para pruebas y prototipos, no para datos de producción.

El trust store por defecto de Java incluye ISRG Root X1, la raíz de Let's Encrypt que usa el broker, en todas las versiones actuales del JDK. No hace falta configurar ningún keystore para TLS.

Ejemplo 1: cliente MQTT 5 con Eclipse Paho

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

public class PahoDemo {

    static String env(String key, String def) {
        String v = System.getenv(key);
        return (v == null || v.isEmpty()) ? def : v;
    }

    public static void main(String[] args) throws Exception {
        String serverUri = env("MQTT_URL", "ssl://broker.connectmq.com:8883");
        String clientId = env("MQTT_CLIENT_ID", "paho-j6t1ve");
        String prefix = env("MQTT_PREFIX", "connectmq/demo/j6t1ve");
        String user = System.getenv("MQTT_USERNAME");
        String commandFilter = prefix + "/cmd/#";
        ExecutorService worker = Executors.newSingleThreadExecutor();

        MqttClient client = new MqttClient(serverUri, clientId, new MemoryPersistence());

        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setAutomaticReconnect(true);
        options.setCleanStart(false);
        options.setSessionExpiryInterval(600L);
        options.setKeepAliveInterval(30);
        if (user != null) {
            options.setUserName(user);
            options.setPassword(env("MQTT_PASSWORD", "").getBytes(StandardCharsets.UTF_8));
        }
        MqttMessage will = new MqttMessage("offline".getBytes(StandardCharsets.UTF_8));
        will.setQos(1);
        will.setRetained(true);
        options.setWill(prefix + "/status", will);

        client.setCallback(new MqttCallback() {
            @Override
            public void connectComplete(boolean reconnect, String uri) {
                System.out.println((reconnect ? "reconnected to " : "connected to ") + uri);
                if (reconnect) {
                    // Nunca bloquees el hilo del callback; vuelve a suscribirte en un worker.
                    worker.submit(() -> {
                        try {
                            client.subscribe(commandFilter, 1);
                        } catch (MqttException e) {
                            System.err.println("resubscribe failed: " + e.getMessage());
                        }
                    });
                }
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                System.out.println("command on " + topic + ": "
                        + new String(message.getPayload(), StandardCharsets.UTF_8));
            }

            @Override
            public void disconnected(MqttDisconnectResponse response) {
                System.out.println("disconnected: " + response.getReasonString());
            }

            @Override
            public void mqttErrorOccurred(MqttException exception) {
                System.err.println("error: " + exception.getMessage());
            }

            @Override
            public void deliveryComplete(IMqttToken token) { }

            @Override
            public void authPacketArrived(int reasonCode, MqttProperties properties) { }
        });

        client.connect(options);
        client.subscribe(commandFilter, 1);

        MqttMessage online = new MqttMessage("online".getBytes(StandardCharsets.UTF_8));
        online.setQos(1);
        online.setRetained(true);
        client.publish(prefix + "/status", online);

        while (true) {
            String json = String.format("{\"device\":\"%s\",\"temperature\":%.2f,\"ts\":\"%s\"}",
                    clientId, 20 + Math.random() * 5, Instant.now());
            MqttMessage msg = new MqttMessage(json.getBytes(StandardCharsets.UTF_8));
            msg.setQos(1);
            try {
                client.publish(prefix + "/telemetry", msg);
            } catch (MqttException e) {
                System.err.println("publish failed: " + e.getMessage());
            }
            Thread.sleep(5000);
        }
    }
}

Puntos clave: el esquema ssl:// activa TLS; setAutomaticReconnect(true) gestiona las caídas; setCleanStart(false) con una expiración de sesión conserva las suscripciones durante cortes breves. Volver a suscribirse en connectComplete cubre los cortes más largos, pero se ejecuta en un hilo worker porque llamar a métodos bloqueantes del cliente desde un callback de Paho puede bloquear el cliente.

Ejemplo 2: HiveMQ MQTT Client

import java.nio.charset.StandardCharsets;
import java.time.Instant;

import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;

public class HiveMqDemo {

    static String env(String key, String def) {
        String v = System.getenv(key);
        return (v == null || v.isEmpty()) ? def : v;
    }

    public static void main(String[] args) throws Exception {
        String host = env("MQTT_HOST", "broker.connectmq.com");
        String clientId = env("MQTT_CLIENT_ID", "hivemq-j6t1ve");
        String prefix = env("MQTT_PREFIX", "connectmq/demo/j6t1ve");
        String user = System.getenv("MQTT_USERNAME");

        Mqtt5AsyncClient client = MqttClient.builder()
                .useMqttVersion5()
                .identifier(clientId)
                .serverHost(host)
                .serverPort(8883)
                .sslWithDefaultConfig()
                .automaticReconnectWithDefaultConfig()
                .addConnectedListener(ctx -> System.out.println("connected"))
                .addDisconnectedListener(ctx ->
                        System.out.println("disconnected: " + ctx.getCause().getMessage()))
                .buildAsync();

        var connect = client.connectWith()
                .cleanStart(false)
                .sessionExpiryInterval(600)
                .keepAlive(30)
                .willPublish()
                    .topic(prefix + "/status")
                    .payload("offline".getBytes(StandardCharsets.UTF_8))
                    .qos(MqttQos.AT_LEAST_ONCE)
                    .retain(true)
                    .applyWillPublish();
        if (user != null) {
            connect = connect.simpleAuth()
                    .username(user)
                    .password(env("MQTT_PASSWORD", "").getBytes(StandardCharsets.UTF_8))
                    .applySimpleAuth();
        }
        connect.send().join();

        client.subscribeWith()
                .topicFilter(prefix + "/cmd/#")
                .qos(MqttQos.AT_LEAST_ONCE)
                .callback(publish -> System.out.println("command on " + publish.getTopic() + ": "
                        + new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8)))
                .send()
                .join();

        client.publishWith()
                .topic(prefix + "/status")
                .payload("online".getBytes(StandardCharsets.UTF_8))
                .qos(MqttQos.AT_LEAST_ONCE)
                .retain(true)
                .send();

        while (true) {
            String json = String.format("{\"device\":\"%s\",\"temperature\":%.2f,\"ts\":\"%s\"}",
                    clientId, 20 + Math.random() * 5, Instant.now());
            client.publishWith()
                    .topic(prefix + "/telemetry")
                    .payload(json.getBytes(StandardCharsets.UTF_8))
                    .qos(MqttQos.AT_LEAST_ONCE)
                    .contentType("application/json")
                    .messageExpiryInterval(60)
                    .send()
                    .whenComplete((result, error) -> {
                        if (error != null) System.err.println("publish failed: " + error.getMessage());
                    });
            Thread.sleep(5000);
        }
    }
}

El builder casi se lee como una frase. sslWithDefaultConfig() activa TLS con el trust store del JDK y automaticReconnectWithDefaultConfig() reconecta con backoff exponencial. Las propiedades de MQTT 5 contentType y messageExpiryInterval indican a los consumidores el formato del payload y permiten al broker descartar las lecturas obsoletas pasado un minuto. La visión general de las funciones de MQTT 5 explica estas propiedades con más detalle.

¿Bloqueante, asíncrono o reactivo?

Ambos ejemplos usan un simple bucle while con Thread.sleep para que el flujo sea fácil de seguir, pero los servicios reales rara vez tienen ese aspecto. El estilo de API que elijas condiciona el resto del código, así que conviene decidirlo pronto.

Las llamadas bloqueantes son las más fáciles de leer: connect, subscribe y publish solo retornan cuando el broker ha respondido. MqttClient de Paho y Mqtt5BlockingClient de HiveMQ funcionan así. Son adecuadas para herramientas de línea de comandos, pruebas y gateways pequeños donde un hilo por conexión es aceptable.

Las llamadas asíncronas retornan de inmediato con un token (IMqttToken en MqttAsyncClient de Paho) o un CompletableFuture (HiveMQ). Son adecuadas para servicios de backend que publican a tasas altas, porque una confirmación lenta nunca bloquea el hilo productor. Encadena el manejo de errores al future en lugar de ignorarlo, como hace el ejemplo de HiveMQ con whenComplete.

Los flujos reactivos, disponibles solo en el cliente de HiveMQ mediante su API RxJava, modelan los mensajes entrantes como un Flowable con contrapresión real. Si tu aplicación ya usa Reactor o RxJava, así evitas acumular mensajes en memoria cuando los consumidores se quedan atrás.

Elijas lo que elijas, mantén el procesamiento de mensajes fuera del hilo de red. Una inserción en base de datos o una llamada HTTP dentro de messageArrived o de un callback de HiveMQ retrasa todos los demás mensajes de esa conexión. Pasa el payload a una cola o a un executor y retorna rápido.

Diseño de topics para servicios Java

Los ejemplos usan tres familias de topics: .../telemetry para lecturas, .../cmd/# para comandos y .../status para la disponibilidad. Separar comandos y datos permite que un consumidor se suscriba a la telemetría con un único comodín sin ver nunca los comandos, y que los permisos se puedan razonar más adelante por familia de topics. Pon los identificadores por los que vayas a filtrar, como planta, línea y dispositivo, en los niveles del topic y no solo en el payload JSON, porque el broker solo puede enrutar por topics.

Prueba de ambos clientes

Ejecuta cualquiera de las dos clases y observa los topics desde una terminal:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-j6t1ve -t 'connectmq/demo/j6t1ve/#' -v
mosquitto_pub -h broker.connectmq.com -p 1883 -i send-j6t1ve -t 'connectmq/demo/j6t1ve/cmd/valve' -m 'close'

Ejecuta solo un cliente a la vez con los ID por defecto, o asigna a cada uno su propio ID; dos clientes no pueden compartir una sesión.

Cambio a un broker privado

Para producción, crea un broker en el portal de ConnectMQ y usa el usuario, la contraseña y el client ID generados. El client ID está vinculado a las credenciales emitidas, por lo que debe coincidir exactamente:

export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_PREFIX=warehouse/zone3/gateway

Paho lee la URI completa de MQTT_URL (ssl://broker.connectmq.com:8883); el ejemplo de HiveMQ lee MQTT_HOST.

Lista de comprobación del cliente MQTT en Java

1. Un client ID único por JVM. Varias instancias con el mismo ID se desconectan entre sí.
2. QoS 1 por defecto. El plan Free admite QoS 0 y 1; QoS 2 requiere un plan de pago. Consulta los niveles de QoS de MQTT explicados.
3. No bloquees los callbacks. Pasa el trabajo a un executor en ambas bibliotecas.
4. Gestiona los fallos de publicación. Decide si descartar, almacenar en búfer o reintentar mientras estás desconectado.
5. Registra los reason codes. Las desconexiones de MQTT 5 incluyen un motivo, por ejemplo 0x97 cuando se supera una cuota.
6. Mantén las credenciales fuera del código, en variables de entorno o en un gestor de secretos.

Solución de problemas

La página oficial del proyecto Paho está en eclipse.dev/paho.

Más tutoriales de backend

Si tu equipo también escribe en Go o Rust, compara con el tutorial de cliente MQTT en Go. Para un repaso más amplio entre lenguajes, lee la guía de bibliotecas cliente MQTT.

Conecta tus servicios Java a ConnectMQ

Un broker MQTT gestionado permite que tus servicios Java se conecten a un único endpoint TLS sin ningún broker que operar. ConnectMQ admite MQTT 3.1.1 y 5.0, así que tanto Paho como el cliente de HiveMQ funcionan sin cambios, y el explorador de topics en vivo del portal muestra exactamente lo que publica tu código. Si todavía estás comparando opciones de alojamiento, la guía de brokers MQTT en la nube repasa las ventajas e inconvenientes. Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito.

Preguntas frecuentes

¿Cuál es la mejor biblioteca cliente MQTT para Java?

Las dos son sólidas. Paho no tiene dependencias y está ampliamente documentada; HiveMQ MQTT Client tiene una API fluida moderna, soporte asíncrono y reactivo, y un desarrollo más activo.

¿Eclipse Paho admite MQTT 5 en Java?

Sí, mediante el artefacto independiente org.eclipse.paho.mqttv5.client. El artefacto anterior org.eclipse.paho.client.mqttv3 solo admite MQTT 3.1.1.

¿Cómo activo TLS en un cliente MQTT en Java?

Con Paho, usa una URI ssl://host:8883. Con el cliente de HiveMQ, llama a sslWithDefaultConfig() y usa el puerto 8883. Ambos se apoyan en el trust store del JDK para los certificados de confianza pública.

¿Los clientes MQTT en Java se reconectan automáticamente?

Sí, si lo activas: setAutomaticReconnect(true) en Paho y automaticReconnectWithDefaultConfig() en el cliente de HiveMQ. Vuelve a suscribirte tras las reconexiones cuando la sesión pueda haber expirado.