MQTT in Java mit dem HiveMQ MQTT Client

Der HiveMQ MQTT Client ist eine Open-Source-Java-Bibliothek für MQTT 3.1.1 und MQTT 5 mit drei API-Stilen: blockierend, asynchron (CompletableFuture) und reaktiv (RxJava, mit optionalem Reactor-Modul). Trotz des Namens funktioniert er mit jedem standardkonformen Broker. In diesem Tutorial verbindest du ihn per TLS, publishst und abonnierst mit jedem API-Stil, nutzt MQTT-5-Features und prüfst alles von außen.

Abhängigkeit hinzufügen

Der Client benötigt Java 8 oder neuer. Mit 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")
}

Mit Maven:

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

Prüfe auf Maven Central, welches 1.3.x-Release aktuell ist. Optionale Features wie WebSocket-Transport und die Reactor-API liegen in separaten Artefakten, damit das Kern-JAR klein bleibt.

Einen Client bauen

Jeder Client beginnt mit MqttClient.builder(). Der Builder ist unveränderlich und fluent: Du wählst Protokollversion, Client-ID, Server und Transport und rufst dann buildBlocking(), buildAsync() oder buildRx() auf.

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 mit dem Truststore der JVM
        .automaticReconnectWithDefaultConfig()  // exponentieller Backoff
        .buildBlocking();

Ein paar Dinge solltest du von Anfang an wissen:

• Client-IDs müssen pro Verbindung eindeutig sein. Zwei Prozesse mit derselben ID werfen sich gegenseitig in einer Schleife raus. Erzeuge eine mit dem Client-ID-Generator oder leite sie von einem stabilen Geräte- oder Instanznamen ab.
• sslWithDefaultConfig() nutzt den Standard-Truststore der JVM, der öffentlichen CAs bereits vertraut. Für eine private CA übergibst du eine TrustManagerFactory über .sslConfig().trustManagerFactory(tmf).applySslConfig().
• Für MQTT 3.1.1 verwendest du useMqttVersion3() — der übrige Code sieht fast gleich aus, nur ohne die MQTT-5-Optionen.

Mit Zugangsdaten verbinden

Die Verbindung wird erst geöffnet, wenn du ein CONNECT sendest. Mit der blockierenden API liefert send() das CONNACK zurück oder wirft eine 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)   // Session eine Stunde offline behalten
            .keepAlive(30)
            .send();
    System.out.println("connected");
} catch (Mqtt5ConnAckException e) {
    System.err.println("rejected: " + e.getMqttMessage().getReasonCode());
}

Lehnt der Broker die Verbindung ab, enthält die Exception den MQTT-5-Reason-Code — zum Beispiel BAD_USER_NAME_OR_PASSWORD oder NOT_AUTHORIZED. Schlag den Wert in der MQTT-Reason-Code-Tabelle nach oder lies CONNACK-Reason-Codes erklärt, um die Bedeutung zu verstehen.

Blockierende API: publishen und empfangen

Die blockierende API ist am leichtesten zu lesen und ideal für Kommandozeilen-Tools und Tests. Um Nachrichten zu empfangen, öffnest du einen Publishes-Stream bevor du abonnierst, damit nichts verloren geht:

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

Da der Stream eine Ressource ist, schließt der try-with-resources-Block ihn sauber. In einem Test ist ein leeres Optional nach dem Timeout ein eindeutiges Fehlersignal.

Asynchrone API: CompletableFuture-Ketten

In Services vermeidet die asynchrone API blockierte Threads. Jede Operation liefert ein CompletableFuture, sodass du Connect, Subscribe und Publish verketten kannst:

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

Mqtt5AsyncClient async = client.toAsync();   // gleiche Verbindung, andere 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");
        });

Der an subscribeWith() übergebene callback läuft auf der Netty-Event-Loop des Clients. Halte ihn kurz: Gib schwere Arbeit — Datenbank-Writes, HTTP-Aufrufe — an deinen eigenen Executor ab, sonst bremst du jede andere Nachricht auf dieser Verbindung aus. Topic-Filter testest du vorab im Topic Matcher.

Reaktive API: Streams mit Backpressure

Mqtt5RxClient stellt Abonnements als RxJava-Flowables bereit. Das hilft, wenn der Consumer langsamer sein kann als der Producer, denn Backpressure wird bis zur MQTT-Flow-Control weitergereicht:

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

Nutzt dein Stack Project Reactor (etwa Spring WebFlux), füge das Artefakt hivemq-mqtt-client-reactor hinzu und arbeite stattdessen mit Flux und Mono.

MQTT-5-Features

Die Builder stellen MQTT-5-Properties direkt bereit, du musst also keine rohen Property-Maps bearbeiten:

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)                    // verwerfen, wenn nach 60 s nicht zugestellt
        .responseTopic("testmqtt/java/orders/replies")
        .correlationData("req-42".getBytes(StandardCharsets.UTF_8))
        .userProperties()
            .add("source", "billing-service")
            .add("schema", "v2")
            .applyUserProperties()
        .send();

Mehr dazu in MQTT User Properties und MQTT-5-Features testen.

Reconnects und Lifecycle-Listener

automaticReconnectWithDefaultConfig() versucht es mit exponentiellem Backoff erneut. Um das zu tunen und Statuswechsel zu loggen, konfigurierst du es explizit:

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

Kombiniere Reconnects mit cleanStart(false) und einem Session-Expiry-Intervall: Der Broker behält dann deine Abonnements und puffert QoS-1/2-Nachrichten, solange der Client weg ist. Dafür ist eine stabile Client-ID unverzichtbar — eine zufällige ID bei jedem Start bedeutet jedes Mal eine neue, leere Session. Beachte, dass der geteilte Test-Endpoint broker.connectmq.com keine Offline-Sessions speichert – führe Tests zur Session-Persistenz daher gegen deine eigene private Broker-Instanz aus.

WebSockets und sauberes Herunterfahren

Manche Netzwerke erlauben nur HTTPS-Traffic. Mit dem Artefakt hivemq-mqtt-client-websocket im Classpath kann derselbe Client MQTT über sichere WebSockets tunneln — nur die Transporteinstellungen ändern sich:

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

Wenn deine Anwendung stoppt, rufe disconnect() auf, statt die JVM einfach zu beenden. Ein sauberes DISCONNECT signalisiert dem Broker, deine Will-Nachricht nicht zu veröffentlichen, und mit MQTT 5 kannst du in diesem Moment sogar die Session Expiry mit disconnectWith().sessionExpiryInterval(0).send() überschreiben, wenn die Session nicht mehr gebraucht wird. In einer Spring- oder Jakarta-EE-Anwendung hängst du das an den Shutdown-Callback des Containers, damit Rolling Deployments keine Geister-Sessions hinterlassen oder falsche "offline"-Alarme auslösen.

Von außen testen

Verlass dich nicht nur auf deinen eigenen Client, um zu wissen, dass alles funktioniert. Öffne den Online-MQTT-Client, verbinde dich mit demselben Broker und abonniere testmqtt/java/#. Führe jedes Beispiel aus und prüfe, ob die Nachrichten mit dem erwarteten Topic, QoS und den MQTT-5-Properties ankommen. Publishe dann aus dem Browser auf testmqtt/java/sensors/lab/temp und sieh zu, wie dein asynchroner Callback sie ausgibt.

Nachrichten deines Java-Clients live verfolgen

Für automatisierte Prüfungen passt die blockierende API gut zu JUnit — siehe MQTT-Integrationstests mit JUnit 5. Wenn du noch eine Bibliothek auswählst, stellt Java-MQTT-Clients im Vergleich diesen Client neben Eclipse Paho.

Häufige Stolperfallen

• Im blockierenden Beispiel kommt nichts an — der publishes()-Stream wurde nach dem Abonnement geöffnet oder der globale Filter passt nicht. Öffne ihn zuerst.
• SSLHandshakeException — meist fehlt eine private CA im Truststore, oder du verbindest dich per TLS mit dem unverschlüsselten Port 1883. Nutze 8883 für TLS.
• Callbacks hängen — langsamer Code im Callback blockiert die Event-Loop. Lagere ihn aus.
• Session nach Neustart verloren — zufällige Client-ID oder cleanStart(true). Nutze eine stabile ID und ein Session-Expiry-Intervall.

Der HiveMQ MQTT Client bietet Java einen modernen Client mit vollständigem MQTT-5-Support im API-Stil deiner Wahl. Beginne mit der blockierenden API, um die Verbindung zu beweisen, wechsle für Produktionscode zu asynchron oder reaktiv und halte während der Entwicklung einen externen Client offen, damit du immer siehst, was tatsächlich über den Broker läuft.

Häufig gestellte Fragen