MQTT em Java com o HiveMQ MQTT Client

O HiveMQ MQTT Client é uma biblioteca Java open source para MQTT 3.1.1 e MQTT 5 com três estilos de API: bloqueante, assíncrona (CompletableFuture) e reativa (RxJava, com um módulo opcional para Reactor). Apesar do nome, ele funciona com qualquer broker que siga o padrão. Neste tutorial você conecta o cliente via TLS, publica e assina com cada estilo de API, usa recursos do MQTT 5 e verifica tudo de fora.

Adicione a dependência

O cliente exige Java 8 ou superior. Com 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")
}

Com Maven:

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

Confira no Maven Central a versão 1.3.x mais recente. Recursos opcionais, como o transporte WebSocket e a API Reactor, ficam em artefatos separados, mantendo o jar principal pequeno.

Crie um cliente

Todo cliente começa com MqttClient.builder(). O builder é imutável e fluente: você escolhe a versão do protocolo, o identificador do cliente, o servidor e o transporte e então chama 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 com o trust store da JVM
        .automaticReconnectWithDefaultConfig()  // backoff exponencial
        .buildBlocking();

Vale saber algumas coisas desde já:

• Identificadores de cliente precisam ser únicos por conexão. Dois processos com o mesmo ID derrubam um ao outro em loop. Gere um com o gerador de client ID ou derive-o de um nome estável de dispositivo ou instância.
• sslWithDefaultConfig() usa o trust store padrão da JVM, que já confia nas CAs públicas. Para uma CA privada, passe um TrustManagerFactory via .sslConfig().trustManagerFactory(tmf).applySslConfig().
• Para MQTT 3.1.1, use useMqttVersion3() — o restante do código fica quase igual, sem as opções do MQTT 5.

Conecte com credenciais

A conexão só é aberta quando você envia um CONNECT. Com a API bloqueante, send() retorna o CONNACK ou lança uma exceção:

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)   // manter a sessão por uma hora offline
            .keepAlive(30)
            .send();
    System.out.println("connected");
} catch (Mqtt5ConnAckException e) {
    System.err.println("rejected: " + e.getMqttMessage().getReasonCode());
}

Se o broker recusar a conexão, a exceção traz o reason code do MQTT 5 — por exemplo BAD_USER_NAME_OR_PASSWORD ou NOT_AUTHORIZED. Procure o valor na tabela de reason codes do MQTT ou leia reason codes do CONNACK explicados para entender cada um.

API bloqueante: publicar e receber

A API bloqueante é a mais fácil de ler e ideal para ferramentas de linha de comando e testes. Para receber mensagens, você abre um stream Publishes antes de assinar, assim nada se perde:

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

Como o stream é um recurso, o bloco try-with-resources o fecha de forma limpa. Em um teste, um Optional vazio após o timeout é um sinal claro de falha.

API assíncrona: cadeias de CompletableFuture

Em serviços, a API assíncrona evita bloquear threads. Cada operação retorna um CompletableFuture, então você pode encadear conexão, assinatura e publicação:

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

Mqtt5AsyncClient async = client.toAsync();   // mesma conexão, outra 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");
        });

O callback passado para subscribeWith() roda no event loop Netty do cliente. Mantenha-o curto: entregue trabalho pesado — gravações em banco, chamadas HTTP — para o seu próprio executor, ou você vai atrasar todas as outras mensagens daquela conexão. Para escolher filtros de topic com segurança, teste-os antes no topic matcher.

API reativa: streams com backpressure

Mqtt5RxClient expõe as assinaturas como Flowables do RxJava. Isso é útil quando o consumidor pode ser mais lento que o produtor, porque o backpressure é propagado para o controle de fluxo do 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);

Se a sua stack usa Project Reactor (Spring WebFlux, por exemplo), adicione o artefato hivemq-mqtt-client-reactor e trabalhe com Flux e Mono.

Recursos do MQTT 5

Os builders expõem as propriedades do MQTT 5 diretamente, sem precisar manipular mapas de propriedades brutos:

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)                    // descartar se não entregue em 60 s
        .responseTopic("testmqtt/java/orders/replies")
        .correlationData("req-42".getBytes(StandardCharsets.UTF_8))
        .userProperties()
            .add("source", "billing-service")
            .add("schema", "v2")
            .applyUserProperties()
        .send();

Saiba mais em user properties do MQTT e testando recursos do MQTT 5.

Reconexões e listeners de ciclo de vida

automaticReconnectWithDefaultConfig() tenta novamente com backoff exponencial. Para ajustar isso e registrar mudanças de estado, configure explicitamente:

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 as reconexões com cleanStart(false) e um intervalo de expiração de sessão: o broker mantém suas assinaturas e enfileira mensagens QoS 1/2 enquanto o cliente está fora. Um identificador de cliente estável é essencial para isso — um ID aleatório a cada inicialização significa uma sessão nova e vazia toda vez. Observe que o endpoint compartilhado de testes broker.connectmq.com não mantém sessões offline; rode os testes de persistência de sessão contra sua própria instância privada de broker.

WebSockets e encerramento limpo

Algumas redes só permitem tráfego HTTPS. Com o artefato hivemq-mqtt-client-websocket no classpath, o mesmo cliente pode tunelar MQTT por WebSockets seguros — só as configurações de transporte mudam:

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

Quando sua aplicação parar, chame disconnect() em vez de simplesmente encerrar a JVM. Um DISCONNECT limpo avisa o broker para não publicar sua mensagem Will e, com MQTT 5, você pode até alterar a expiração da sessão nesse momento com disconnectWith().sessionExpiryInterval(0).send() quando a sessão não for mais necessária. Em uma aplicação Spring ou Jakarta EE, conecte isso ao callback de shutdown do container para que deploys graduais não deixem sessões fantasmas nem disparem alertas falsos de "offline".

Teste de fora

Não confie apenas no seu próprio cliente para dizer que tudo funciona. Abra o cliente MQTT online, conecte-se ao mesmo broker e assine testmqtt/java/#. Execute cada exemplo e confirme que as mensagens chegam com o topic, o QoS e as propriedades do MQTT 5 esperados. Depois publique pelo navegador em testmqtt/java/sensors/lab/temp e veja seu callback assíncrono imprimir a mensagem.

Veja ao vivo as mensagens do seu cliente Java

Para verificações automatizadas, a API bloqueante combina bem com JUnit — veja testes de integração MQTT com JUnit 5. Se você ainda está escolhendo uma biblioteca, clientes MQTT para Java comparados coloca este cliente lado a lado com o Eclipse Paho.

Armadilhas comuns

• Nada recebido no exemplo bloqueante — o stream publishes() foi aberto depois da assinatura, ou o filtro global não corresponde. Abra-o primeiro.
• SSLHandshakeException — geralmente uma CA privada ausente no trust store, ou conexão TLS na porta sem criptografia 1883. Use a 8883 para TLS.
• Callbacks travam — código lento dentro do callback bloqueia o event loop. Delegue-o.
• Sessão perdida após reiniciar — client ID aleatório ou cleanStart(true). Use um ID estável e um intervalo de expiração de sessão.

O HiveMQ MQTT Client oferece ao Java um cliente moderno e totalmente compatível com MQTT 5 no estilo de API que você preferir. Comece com a API bloqueante para provar a conexão, migre para assíncrona ou reativa no código de produção e mantenha um cliente externo aberto durante o desenvolvimento para sempre ver o que realmente passa pelo broker.

Perguntas frequentes