MQTT in Java with the HiveMQ MQTT Client

The HiveMQ MQTT Client is an open-source Java library for MQTT 3.1.1 and MQTT 5 with three API styles: blocking, asynchronous (CompletableFuture) and reactive (RxJava, with an optional Reactor module). Despite the name, it works with any standards-compliant broker. This tutorial connects it over TLS, publishes and subscribes with each API style, uses MQTT 5 features and shows how to verify everything from the outside.

Add the dependency

The client needs Java 8 or later. With 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")
}

With Maven:

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

Check Maven Central for the newest 1.3.x release. Optional features such as WebSocket transport and the Reactor API live in separate artifacts, so the core jar stays small.

Build a client

Every client starts from MqttClient.builder(). The builder is immutable and fluent: you choose the protocol version, the client identifier, the server and the transport, then call buildBlocking(), buildAsync() or 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 with the JVM trust store
        .automaticReconnectWithDefaultConfig()  // exponential backoff
        .buildBlocking();

A few things are worth knowing up front:

• Client identifiers must be unique per connection. Two processes with the same ID kick each other off in a loop. Generate one with the client ID generator or derive it from a stable device or instance name.
• sslWithDefaultConfig() uses the JVM's default trust store, which already trusts public CAs. For a private CA, pass a TrustManagerFactory via .sslConfig().trustManagerFactory(tmf).applySslConfig().
• For MQTT 3.1.1, use useMqttVersion3() — the rest of the code looks almost the same, minus the MQTT 5 options.

Connect with credentials

The connection is not opened until you send a CONNECT. With the blocking API, send() returns the CONNACK or throws:

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)   // keep the session for one hour offline
            .keepAlive(30)
            .send();
    System.out.println("connected");
} catch (Mqtt5ConnAckException e) {
    System.err.println("rejected: " + e.getMqttMessage().getReasonCode());
}

If the broker refuses the connection, the exception carries the MQTT 5 reason code — for example BAD_USER_NAME_OR_PASSWORD or NOT_AUTHORIZED. Look the value up in the MQTT reason codes table or read CONNACK reason codes explained to see what each one means.

Blocking API: publish and receive

The blocking API is the easiest to read and ideal for command-line tools and tests. To receive messages you open a Publishes stream before subscribing, so nothing is missed:

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

Because the stream is a resource, the try-with-resources block closes it cleanly. In a test, an empty Optional after the timeout is a clear failure signal.

Async API: CompletableFuture chains

For services, the async API avoids blocking threads. Each operation returns a CompletableFuture, so you can chain connect, subscribe and publish:

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

Mqtt5AsyncClient async = client.toAsync();   // same connection, different 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");
        });

The callback passed to subscribeWith() runs on the client's Netty event loop. Keep it short: hand heavy work — database writes, HTTP calls — to your own executor, or you will slow down every other message on that connection. To pick topic filters with confidence, try them first in the topic matcher.

Reactive API: streams with backpressure

Mqtt5RxClient exposes subscriptions as RxJava Flowables. That is useful when the consumer can be slower than the producer, because backpressure is propagated to MQTT flow control:

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

If your stack uses Project Reactor (Spring WebFlux, for example), add the hivemq-mqtt-client-reactor artifact and work with Flux and Mono instead.

MQTT 5 features

The builders expose MQTT 5 properties directly, so there is no need to manipulate raw property maps:

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

Read more in MQTT user properties and testing MQTT 5 features.

Reconnects and lifecycle listeners

automaticReconnectWithDefaultConfig() retries with exponential backoff. To tune it and to log state changes, configure it explicitly:

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

Combine reconnects with cleanStart(false) and a session expiry interval: the broker then keeps your subscriptions and queues QoS 1/2 messages while the client is away. A stable client identifier is essential for that — a random ID on every start means a fresh, empty session each time. Note that the shared test endpoint broker.connectmq.com does not keep offline sessions, so run session-persistence tests against your own private broker instance.

WebSockets and clean shutdown

Some networks only allow HTTPS traffic. With the hivemq-mqtt-client-websocket artifact on the classpath, the same client can tunnel MQTT through secure WebSockets — only the transport settings change:

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

When your application stops, call disconnect() instead of just exiting the JVM. A clean DISCONNECT tells the broker not to publish your Will message, and with MQTT 5 you can even override the session expiry at that moment with disconnectWith().sessionExpiryInterval(0).send() when the session is no longer needed. In a Spring or Jakarta EE application, hook this into the container's shutdown callback so rolling deployments do not leave ghost sessions or trigger false "offline" alerts.

Test it from the outside

Do not trust your own client alone to tell you that it works. Open the online MQTT client, connect to the same broker and subscribe to testmqtt/java/#. Run each example and confirm the messages arrive with the expected topic, QoS and MQTT 5 properties. Then publish from the browser to testmqtt/java/sensors/lab/temp and watch your async callback print it.

Watch your Java client's messages live

For automated checks, the blocking API pairs well with JUnit — see MQTT integration tests with JUnit 5. If you are still choosing a library, Java MQTT clients compared puts this client next to Eclipse Paho.

Common pitfalls

• Nothing received in the blocking example — the publishes() stream was opened after the subscription, or the global filter does not match. Open it first.
• SSLHandshakeException — usually a private CA missing from the trust store, or connecting with TLS to the plain port 1883. Use 8883 for TLS.
• Callbacks stall — slow code inside the callback blocks the event loop. Offload it.
• Session lost after restart — random client ID or cleanStart(true). Use a stable ID and a session expiry interval.

The HiveMQ MQTT Client gives Java a modern, fully MQTT 5 client with the API style you prefer. Start with the blocking API to prove the connection, move to async or reactive for production code, and keep an external client open while you develop so you always see what actually crosses the broker.

Frequently asked questions