MQTT en Java avec le HiveMQ MQTT Client

Le HiveMQ MQTT Client est une bibliothèque Java open source pour MQTT 3.1.1 et MQTT 5, avec trois styles d'API : bloquant, asynchrone (CompletableFuture) et réactif (RxJava, avec un module Reactor optionnel). Malgré son nom, il fonctionne avec n'importe quel broker conforme au standard. Dans ce tutoriel, vous le connectez en TLS, publiez et vous abonnez avec chaque style d'API, utilisez les fonctionnalités MQTT 5 et vérifiez tout depuis l'extérieur.

Ajouter la dépendance

Le client nécessite Java 8 ou supérieur. Avec 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")
}

Avec Maven :

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

Vérifiez sur Maven Central la dernière version 1.3.x. Les fonctionnalités optionnelles comme le transport WebSocket et l'API Reactor sont dans des artefacts séparés, ce qui garde le jar principal léger.

Construire un client

Tout client commence par MqttClient.builder(). Le builder est immuable et fluide : vous choisissez la version du protocole, l'identifiant client, le serveur et le transport, puis appelez buildBlocking(), buildAsync() ou 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 avec le trust store de la JVM
        .automaticReconnectWithDefaultConfig()  // backoff exponentiel
        .buildBlocking();

Quelques points à connaître d'emblée :

• Les identifiants client doivent être uniques par connexion. Deux processus avec le même ID s'éjectent mutuellement en boucle. Générez-en un avec le générateur de client ID ou dérivez-le d'un nom stable d'appareil ou d'instance.
• sslWithDefaultConfig() utilise le trust store par défaut de la JVM, qui fait déjà confiance aux CA publiques. Pour une CA privée, passez une TrustManagerFactory via .sslConfig().trustManagerFactory(tmf).applySslConfig().
• Pour MQTT 3.1.1, utilisez useMqttVersion3() : le reste du code est presque identique, sans les options MQTT 5.

Se connecter avec des identifiants

La connexion n'est ouverte qu'à l'envoi d'un CONNECT. Avec l'API bloquante, send() renvoie le CONNACK ou lève une exception :

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)   // conserver la session une heure hors ligne
            .keepAlive(30)
            .send();
    System.out.println("connected");
} catch (Mqtt5ConnAckException e) {
    System.err.println("rejected: " + e.getMqttMessage().getReasonCode());
}

Si le broker refuse la connexion, l'exception contient le reason code MQTT 5, par exemple BAD_USER_NAME_OR_PASSWORD ou NOT_AUTHORIZED. Cherchez la valeur dans la table des reason codes MQTT ou lisez les reason codes CONNACK expliqués pour comprendre chacun d'eux.

API bloquante : publier et recevoir

L'API bloquante est la plus lisible et idéale pour les outils en ligne de commande et les tests. Pour recevoir des messages, ouvrez un flux Publishes avant de vous abonner, afin de ne rien manquer :

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();

Le flux étant une ressource, le bloc try-with-resources le ferme proprement. Dans un test, un Optional vide après le timeout est un signal d'échec clair.

API asynchrone : chaînes de CompletableFuture

Dans les services, l'API asynchrone évite de bloquer des threads. Chaque opération renvoie un CompletableFuture, vous pouvez donc enchaîner connexion, abonnement et publication :

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

Mqtt5AsyncClient async = client.toAsync();   // même connexion, autre 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");
        });

Le callback passé à subscribeWith() s'exécute sur l'event loop Netty du client. Gardez-le court : confiez le travail lourd (écritures en base, appels HTTP) à votre propre executor, sinon vous ralentirez tous les autres messages de cette connexion. Pour choisir vos filtres de topic en toute confiance, testez-les d'abord dans le topic matcher.

API réactive : flux avec backpressure

Mqtt5RxClient expose les abonnements sous forme de Flowable RxJava. C'est utile quand le consommateur peut être plus lent que le producteur, car le backpressure est propagé jusqu'au contrôle de flux 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 votre stack utilise Project Reactor (Spring WebFlux, par exemple), ajoutez l'artefact hivemq-mqtt-client-reactor et travaillez avec Flux et Mono.

Fonctionnalités MQTT 5

Les builders exposent directement les propriétés MQTT 5, sans manipuler de maps de propriétés brutes :

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)                    // abandonner si non livré après 60 s
        .responseTopic("testmqtt/java/orders/replies")
        .correlationData("req-42".getBytes(StandardCharsets.UTF_8))
        .userProperties()
            .add("source", "billing-service")
            .add("schema", "v2")
            .applyUserProperties()
        .send();

Pour aller plus loin : user properties MQTT et tester les fonctionnalités MQTT 5.

Reconnexions et listeners de cycle de vie

automaticReconnectWithDefaultConfig() réessaie avec un backoff exponentiel. Pour l'ajuster et journaliser les changements d'état, configurez-le explicitement :

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();

Combinez les reconnexions avec cleanStart(false) et un intervalle d'expiration de session : le broker conserve alors vos abonnements et met en file les messages QoS 1/2 pendant l'absence du client. Un identifiant client stable est indispensable : un ID aléatoire à chaque démarrage signifie une session neuve et vide à chaque fois. Notez que l'endpoint de test partagé broker.connectmq.com ne conserve pas les sessions hors ligne : exécutez les tests de persistance de session sur votre propre instance de broker privée.

WebSockets et arrêt propre

Certains réseaux n'autorisent que le trafic HTTPS. Avec l'artefact hivemq-mqtt-client-websocket dans le classpath, le même client peut faire passer MQTT par des WebSockets sécurisés ; seuls les paramètres de transport changent :

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

Lorsque votre application s'arrête, appelez disconnect() au lieu de simplement quitter la JVM. Un DISCONNECT propre indique au broker de ne pas publier votre message Will et, avec MQTT 5, vous pouvez même modifier l'expiration de session à ce moment avec disconnectWith().sessionExpiryInterval(0).send() quand la session n'est plus utile. Dans une application Spring ou Jakarta EE, branchez cela sur le callback d'arrêt du conteneur pour que les déploiements progressifs ne laissent pas de sessions fantômes ni ne déclenchent de fausses alertes « offline ».

Tester depuis l'extérieur

Ne vous fiez pas uniquement à votre propre client pour savoir que tout fonctionne. Ouvrez le client MQTT en ligne, connectez-vous au même broker et abonnez-vous à testmqtt/java/#. Exécutez chaque exemple et vérifiez que les messages arrivent avec le topic, le QoS et les propriétés MQTT 5 attendus. Publiez ensuite depuis le navigateur sur testmqtt/java/sensors/lab/temp et regardez votre callback asynchrone l'afficher.

Suivez en direct les messages de votre client Java

Pour des vérifications automatisées, l'API bloquante se marie bien avec JUnit : voir tests d'intégration MQTT avec JUnit 5. Si vous choisissez encore une bibliothèque, les clients MQTT Java comparés place ce client à côté d'Eclipse Paho.

Pièges courants

• Rien n'est reçu dans l'exemple bloquant : le flux publishes() a été ouvert après l'abonnement, ou le filtre global ne correspond pas. Ouvrez-le en premier.
• SSLHandshakeException : en général une CA privée absente du trust store, ou une connexion TLS sur le port en clair 1883. Utilisez 8883 pour TLS.
• Les callbacks se figent : du code lent dans le callback bloque l'event loop. Déportez-le.
• Session perdue après redémarrage : client ID aléatoire ou cleanStart(true). Utilisez un ID stable et un intervalle d'expiration de session.

Le HiveMQ MQTT Client apporte à Java un client moderne, entièrement compatible MQTT 5, dans le style d'API de votre choix. Commencez par l'API bloquante pour valider la connexion, passez à l'asynchrone ou au réactif pour le code de production, et gardez un client externe ouvert pendant le développement pour toujours voir ce qui transite réellement par le broker.

Questions fréquentes