Java MQTT Client Tutorial: Eclipse Paho vs. HiveMQ Client mit TLS und MQTT 5

Bei der Wahl eines Java MQTT Clients läuft es meist auf zwei Bibliotheken hinaus: Eclipse Paho und den HiveMQ MQTT Client. Dieses Tutorial vergleicht beide ehrlich und implementiert anschließend mit jeder dasselbe Publish/Subscribe-Programm – mit MQTT 5 über TLS auf Port 8883, automatischen Reconnects und einem Last Will –, damit Sie die passende Bibliothek für Ihre Codebasis wählen können.

Java-MQTT-Client-Bibliotheken im Vergleich

Beide Bibliotheken sind Open Source, beide sprechen MQTT 5 und beide werden produktiv eingesetzt. Sie unterscheiden sich vor allem im API-Stil und darin, wie aktiv sie weiterentwickelt werden.

Eine praktische Faustregel: Paho ist klein, frei von Abhängigkeiten und aus unzähligen Beispielen bekannt – attraktiv für Android oder eingeschränkte Umgebungen. Der HiveMQ-Client hat eine modernere API und kommt mit Back-Pressure bei hohen Nachrichtenraten gut zurecht, was zu Backend-Services passt. Spring-Entwickler nutzen Paho oft indirekt über Spring Integration; diesen Weg beschreibt der Spring-Boot-MQTT-Leitfaden.

Projekt-Setup

Beide Beispiele verwenden Java 17 und Maven. Fügen Sie eine oder beide Abhängigkeiten in die pom.xml ein:

<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>

Zum Testen verbinden Sie sich mit dem öffentlichen MQTT-Broker unter broker.connectmq.com. Er akzeptiert anonyme Clients mit jeder nicht leeren, eindeutigen Client ID. Verwenden Sie ein zufälliges Präfix wie connectmq/demo/j6t1ve und denken Sie daran, dass öffentliche Topics von jedem gelesen werden können: Der Broker ist für Tests und Prototypen gedacht, nicht für Produktionsdaten.

Der Standard-Truststore von Java enthält in allen aktuellen JDK-Versionen ISRG Root X1, die Let's-Encrypt-Root, die der Broker verwendet. Für TLS ist keine Keystore-Einrichtung nötig.

Beispiel 1: Eclipse Paho MQTT 5 Client

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) {
                    // Callback-Thread nie blockieren; Resubscribe auf einem 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);
        }
    }
}

Die wichtigsten Punkte: Das Schema ssl:// aktiviert TLS; setAutomaticReconnect(true) fängt Verbindungsabbrüche ab; setCleanStart(false) mit einer Session Expiry erhält Subscriptions über kurze Ausfälle hinweg. Das Resubscribe in connectComplete deckt längere Ausfälle ab, läuft aber auf einem Worker-Thread, weil blockierende Client-Methoden aus einem Paho-Callback heraus den Client lahmlegen können.

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

Der Builder liest sich fast wie ein Satz. sslWithDefaultConfig() aktiviert TLS mit dem JDK-Truststore, und automaticReconnectWithDefaultConfig() verbindet sich mit exponentiellem Back-off neu. Die MQTT-5-Properties contentType und messageExpiryInterval teilen Konsumenten das Payload-Format mit und lassen den Broker veraltete Messwerte nach einer Minute verwerfen. Die Übersicht der MQTT-5-Funktionen erklärt diese Properties ausführlicher.

Blockierend, asynchron oder reaktiv?

Beide Beispiele verwenden eine einfache while-Schleife mit Thread.sleep, damit der Ablauf leicht nachvollziehbar ist – echte Services sehen aber selten so aus. Der gewählte API-Stil prägt den restlichen Code, deshalb lohnt sich eine frühe Entscheidung.

Blockierende Aufrufe sind am leichtesten zu lesen: Connect, Subscribe und Publish kehren jeweils erst zurück, wenn der Broker geantwortet hat. Pahos MqttClient und der HiveMQ-Mqtt5BlockingClient arbeiten so. Sie eignen sich für Kommandozeilen-Tools, Tests und kleine Gateways, bei denen ein Thread pro Verbindung vertretbar ist.

Asynchrone Aufrufe kehren sofort mit einem Token (IMqttToken in Pahos MqttAsyncClient) oder einer CompletableFuture (HiveMQ) zurück. Sie eignen sich für Backend-Services mit hohen Publish-Raten, weil eine langsame Bestätigung nie den Producer-Thread blockiert. Hängen Sie die Fehlerbehandlung an die Future, statt sie zu ignorieren – so wie das HiveMQ-Beispiel mit whenComplete.

Reaktive Streams, die nur der HiveMQ-Client über seine RxJava-API bietet, modellieren eingehende Nachrichten als Flowable mit echter Back-Pressure. Nutzt Ihre Anwendung bereits Reactor oder RxJava, vermeiden Sie so, dass sich Nachrichten im Speicher stauen, wenn Konsumenten nicht hinterherkommen.

Wofür Sie sich auch entscheiden: Halten Sie die Nachrichtenverarbeitung vom Netzwerk-Thread fern. Ein Datenbank-Insert oder HTTP-Aufruf in messageArrived oder einem HiveMQ-Callback verzögert jede andere Nachricht auf dieser Verbindung. Schieben Sie den Payload in eine Queue oder einen Executor und kehren Sie schnell zurück.

Topic-Design für Java-Services

Die Beispiele nutzen drei Topic-Familien: .../telemetry für Messwerte, .../cmd/# für Befehle und .../status für die Verfügbarkeit. Wenn Befehle und Daten getrennt sind, kann ein Konsument Telemetrie mit einer einzigen Wildcard abonnieren, ohne je Befehle zu sehen, und Berechtigungen lassen sich später pro Topic-Familie festlegen. Legen Sie Kennungen, nach denen Sie filtern wollen – etwa Standort, Linie und Gerät –, in Topic-Ebenen ab und nicht nur im JSON-Payload, denn der Broker kann nur anhand von Topics routen. In Industrie-4.0-Projekten folgen diese Ebenen meist der ISA-95-Hierarchie eines Unified Namespace (UNS), etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur, sodass ein Java-Service im MES- oder ERP-Umfeld eine ganze Linie mit einer Wildcard abonnieren kann.

Beide Clients testen

Starten Sie eine der beiden Klassen und beobachten Sie die Topics in einem 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'

Starten Sie mit den Standard-IDs immer nur einen Client gleichzeitig oder geben Sie jedem eine eigene ID; zwei Clients können sich keine Session teilen.

Wechsel zu einem privaten Broker

Für den Produktivbetrieb erstellen Sie einen Broker im ConnectMQ-Portal und verwenden den generierten Benutzernamen, das Passwort und die Client ID. Die Client ID ist an die ausgestellten Zugangsdaten gebunden und muss daher exakt übereinstimmen:

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

Paho liest die vollständige URI aus MQTT_URL (ssl://broker.connectmq.com:8883); das HiveMQ-Beispiel liest MQTT_HOST.

Checkliste für Java MQTT Clients

1. Eindeutige Client ID pro JVM. Mehrere Instanzen mit derselben ID werfen sich gegenseitig vom Broker.
2. Standardmäßig QoS 1. Der Free-Plan unterstützt QoS 0 und 1; QoS 2 erfordert einen kostenpflichtigen Plan. Siehe MQTT-QoS-Stufen erklärt.
3. Callbacks nicht blockieren. Geben Sie Arbeit in beiden Bibliotheken an einen Executor ab.
4. Publish-Fehler behandeln. Legen Sie fest, ob während einer Trennung verworfen, gepuffert oder erneut gesendet wird.
5. Reason Codes protokollieren. MQTT-5-Disconnects enthalten einen Grund, zum Beispiel 0x97 bei überschrittener Quota.
6. Zugangsdaten außerhalb des Codes halten, in Umgebungsvariablen oder einem Secret Store.

Fehlerbehebung

Die offizielle Paho-Projektseite finden Sie unter eclipse.dev/paho.

Weitere Backend-Tutorials

Schreibt Ihr Team auch Go oder Rust, vergleichen Sie mit dem Go-MQTT-Client-Tutorial. Einen breiteren Überblick über alle Sprachen bietet der Leitfaden zu MQTT-Client-Bibliotheken.

Verbinden Sie Ihre Java-Services mit ConnectMQ

Mit einem verwalteten MQTT-Broker verbinden sich Ihre Java-Services über einen einzigen TLS-Endpunkt, ohne dass Sie einen Broker betreiben müssen. ConnectMQ unterstützt MQTT 3.1.1 und 5.0, sodass Paho und der HiveMQ-Client unverändert funktionieren, und der Live-Topic-Explorer im Portal zeigt genau, was Ihr Code veröffentlicht. Vergleichen Sie noch Hosting-Optionen, beleuchtet der Cloud-MQTT-Broker-Leitfaden die Vor- und Nachteile. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker, ganz ohne Kreditkarte.

Häufig gestellte Fragen

Welche Java-MQTT-Client-Bibliothek ist die beste?

Beide sind solide. Paho kommt ohne Abhängigkeiten aus und ist breit dokumentiert; der HiveMQ MQTT Client bietet eine moderne Fluent-API, Unterstützung für asynchrone und reaktive Programmierung sowie eine aktivere Weiterentwicklung.

Unterstützt Eclipse Paho MQTT 5 in Java?

Ja, über das separate Artefakt org.eclipse.paho.mqttv5.client. Das ältere Artefakt org.eclipse.paho.client.mqttv3 unterstützt nur MQTT 3.1.1.

Wie aktiviere ich TLS in einem Java MQTT Client?

Bei Paho verwenden Sie eine URI der Form ssl://host:8883. Beim HiveMQ-Client rufen Sie sslWithDefaultConfig() auf und nutzen Port 8883. Beide verlassen sich bei öffentlich vertrauenswürdigen Zertifikaten auf den JDK-Truststore.

Verbinden sich Java MQTT Clients automatisch neu?

Ja, sofern aktiviert: setAutomaticReconnect(true) bei Paho und automaticReconnectWithDefaultConfig() beim HiveMQ-Client. Abonnieren Sie nach Reconnects erneut, wenn die Session abgelaufen sein könnte.