Tutoriel client MQTT Java : Eclipse Paho vs HiveMQ Client avec TLS et MQTT 5

Choisir un client MQTT Java revient généralement à trancher entre deux bibliothèques : Eclipse Paho et HiveMQ MQTT Client. Ce tutoriel les compare honnêtement, puis implémente le même programme publish/subscribe avec chacune, en MQTT 5 sur TLS (port 8883), avec reconnexion automatique et Last Will, afin que vous puissiez choisir celle qui convient le mieux à votre code.

Comparatif des bibliothèques client MQTT Java

Les deux bibliothèques sont open source, prennent en charge MQTT 5 et sont utilisées en production. Elles diffèrent surtout par le style de leur API et par le dynamisme de leur développement.

Règle pratique : Paho est léger, sans dépendance et familier grâce à d'innombrables exemples, ce qui le rend intéressant pour Android ou les environnements contraints. Le client HiveMQ propose une API plus moderne et gère bien la contre-pression (back-pressure) à des débits de messages élevés, ce qui convient aux services backend. Les développeurs Spring utilisent souvent Paho indirectement via Spring Integration ; consultez le guide MQTT avec Spring Boot pour cette approche.

Configuration du projet

Les deux exemples utilisent Java 17 et Maven. Ajoutez l'une des dépendances, ou les deux, dans 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>

Pour vos tests, connectez-vous au serveur MQTT public gratuit sur broker.connectmq.com. Il accepte les clients anonymes avec n'importe quel client ID unique et non vide. Utilisez un préfixe aléatoire comme connectmq/demo/j6t1ve et gardez à l'esprit que les topics publics peuvent être lus par tout le monde.

Le trust store par défaut de Java inclut ISRG Root X1, la racine Let's Encrypt utilisée par le broker, sur toutes les versions actuelles du JDK. Aucune configuration de keystore n'est nécessaire pour TLS.

Exemple 1 : client Eclipse Paho MQTT 5

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) {
                    // Never block the callback thread; resubscribe on a 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);
        }
    }
}

Points clés : le schéma ssl:// active TLS ; setAutomaticReconnect(true) gère les coupures ; setCleanStart(false) associé à une expiration de session conserve les abonnements lors de courtes interruptions. Le réabonnement dans connectComplete couvre les plus longues, mais il s'exécute sur un thread de travail, car appeler des méthodes bloquantes du client depuis un callback Paho peut figer le client.

Exemple 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);
        }
    }
}

Le builder se lit presque comme une phrase. sslWithDefaultConfig() active TLS avec le trust store du JDK, et automaticReconnectWithDefaultConfig() se reconnecte avec un délai exponentiel (back-off). Les propriétés MQTT 5 contentType et messageExpiryInterval indiquent aux consommateurs le format du payload et permettent au broker d'écarter les relevés obsolètes au bout d'une minute. La présentation des fonctionnalités de MQTT 5 détaille ces propriétés.

Bloquant, asynchrone ou réactif ?

Les deux exemples utilisent une simple boucle while avec Thread.sleep pour que le déroulement soit facile à suivre, mais les vrais services ressemblent rarement à cela. Le style d'API choisi façonne le reste du code : mieux vaut donc trancher tôt.

Les appels bloquants sont les plus lisibles : connect, subscribe et publish ne rendent la main qu'une fois que le broker a répondu. Le MqttClient de Paho et le Mqtt5BlockingClient de HiveMQ fonctionnent ainsi. Ils conviennent aux outils en ligne de commande, aux tests et aux petites passerelles où un thread par connexion est acceptable.

Les appels asynchrones rendent la main immédiatement avec un token (IMqttToken dans le MqttAsyncClient de Paho) ou un CompletableFuture (HiveMQ). Ils conviennent aux services backend qui publient à haut débit, car un acquittement lent ne bloque jamais le thread producteur. Chaînez la gestion des erreurs sur le future au lieu de l'ignorer, comme le fait l'exemple HiveMQ avec whenComplete.

Les flux réactifs, disponibles uniquement dans le client HiveMQ via son API RxJava, modélisent les messages entrants comme un Flowable avec une véritable contre-pression. Si votre application utilise déjà Reactor ou RxJava, cela évite de mettre les messages en mémoire tampon lorsque les consommateurs prennent du retard.

Quel que soit votre choix, sortez le traitement des messages du thread réseau. Une insertion en base de données ou un appel HTTP dans messageArrived ou dans un callback HiveMQ retarde tous les autres messages de cette connexion. Placez le payload dans une file ou un executor et rendez la main rapidement.

Conception des topics pour les services Java

Les exemples utilisent trois familles de topics : .../telemetry pour les relevés, .../cmd/# pour les commandes et .../status pour la disponibilité. Séparer commandes et données permet à un consommateur de s'abonner à la télémétrie avec un seul wildcard sans jamais voir les commandes, et facilite plus tard le raisonnement sur les permissions par famille de topics. Placez les identifiants sur lesquels vous filtrerez, comme le site, la ligne et l'appareil, dans les niveaux du topic plutôt que seulement dans le payload JSON, car le broker ne peut router que sur les topics.

Tester les deux clients

Lancez l'une ou l'autre classe et observez les topics depuis un 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'

N'exécutez qu'un seul client à la fois avec les ID par défaut, ou attribuez à chacun son propre ID ; deux clients ne peuvent pas partager une session.

Passer à un broker privé

En production, créez un broker dans le portail ConnectMQ et utilisez le nom d'utilisateur, le mot de passe et le client ID générés. Le client ID est lié aux identifiants émis et doit donc correspondre exactement :

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

Paho lit l'URI complète depuis MQTT_URL (ssl://broker.connectmq.com:8883) ; l'exemple HiveMQ lit MQTT_HOST.

Check-list du client MQTT Java

1. Un client ID unique par JVM. Plusieurs instances avec le même ID se déconnectent mutuellement.
2. QoS 1 par défaut. Le plan Free prend en charge QoS 0 et 1 ; QoS 2 nécessite un plan payant. Voir les niveaux de QoS MQTT expliqués.
3. Ne bloquez pas les callbacks. Confiez le travail à un executor, quelle que soit la bibliothèque.
4. Gérez les échecs de publication. Décidez s'il faut abandonner, mettre en tampon ou réessayer pendant la déconnexion.
5. Journalisez les reason codes. Les déconnexions MQTT 5 portent une raison, par exemple 0x97 lorsqu'un quota est dépassé.
6. Gardez les identifiants hors du code, dans des variables d'environnement ou un gestionnaire de secrets.

Dépannage

La page officielle du projet Paho se trouve sur eclipse.dev/paho.

Autres tutoriels backend

Si votre équipe écrit aussi du Go ou du Rust, comparez avec le tutoriel client MQTT en Go. Pour un panorama plus large couvrant plusieurs langages, lisez le guide des bibliothèques client MQTT.

Connectez vos services Java à ConnectMQ

Un broker MQTT managé permet à vos services Java de se connecter via un seul endpoint TLS, sans broker à exploiter. ConnectMQ prend en charge MQTT 3.1.1 et 5.0 : Paho comme le client HiveMQ fonctionnent donc sans modification, et l'explorateur de topics en direct du portail montre exactement ce que publie votre code. Si vous comparez encore les options d'hébergement, le guide du broker MQTT cloud en présente les compromis. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire.

Questions fréquentes

Quelle est la meilleure bibliothèque client MQTT Java ?

Les deux sont solides. Paho est sans dépendance et largement documenté ; HiveMQ MQTT Client offre une API fluide moderne, le support asynchrone et réactif, et un développement plus actif.

Eclipse Paho prend-il en charge MQTT 5 en Java ?

Oui, via l'artefact distinct org.eclipse.paho.mqttv5.client. L'ancien artefact org.eclipse.paho.client.mqttv3 ne prend en charge que MQTT 3.1.1.

Comment activer TLS dans un client MQTT Java ?

Avec Paho, utilisez une URI ssl://host:8883. Avec le client HiveMQ, appelez sslWithDefaultConfig() et utilisez le port 8883. Les deux s'appuient sur le trust store du JDK pour les certificats reconnus publiquement.

Les clients MQTT Java se reconnectent-ils automatiquement ?

Oui, si l'option est activée : setAutomaticReconnect(true) dans Paho et automaticReconnectWithDefaultConfig() dans le client HiveMQ. Réabonnez-vous après une reconnexion lorsque la session a pu expirer.