Tutorial de Cliente MQTT em Java: Eclipse Paho vs HiveMQ Client com TLS e MQTT 5

Escolher um cliente MQTT Java geralmente se resume a duas bibliotecas: o Eclipse Paho e o HiveMQ MQTT Client. Este tutorial compara as duas com honestidade e depois implementa o mesmo programa de publish/subscribe com cada uma, usando MQTT 5 via TLS na porta 8883, reconexão automática e um Last Will, para que você escolha a que melhor se adapta ao seu código.

Bibliotecas de cliente MQTT para Java comparadas

As duas bibliotecas são open source, ambas falam MQTT 5 e ambas são usadas em produção. Elas diferem principalmente no estilo da API e no ritmo de desenvolvimento.

Uma regra prática: o Paho é pequeno, sem dependências e familiar graças a inúmeros exemplos, o que o torna atraente para Android ou ambientes restritos. O cliente HiveMQ tem uma API mais moderna e lida bem com back-pressure em altas taxas de mensagens, o que combina com serviços de backend. Desenvolvedores Spring costumam usar o Paho indiretamente por meio do Spring Integration; veja o guia de MQTT com Spring Boot para esse caminho.

Configuração do projeto

Os dois exemplos usam Java 17 e Maven. Adicione uma ou ambas as dependências ao 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>

Para testes, conecte-se ao servidor MQTT público (gratuito) em broker.connectmq.com, adequado para protótipos, nunca para dados de produção. Ele aceita clientes anônimos com qualquer client ID não vazio e único. Use um prefixo aleatório como connectmq/demo/j6t1ve e tenha em mente que os topics públicos podem ser lidos por qualquer pessoa.

O trust store padrão do Java inclui o ISRG Root X1, a raiz da Let's Encrypt usada pelo broker, em todas as versões atuais do JDK. Nenhuma configuração de keystore é necessária para o TLS.

Exemplo 1: cliente MQTT 5 com Eclipse Paho

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) {
                    // Nunca bloqueie a thread do callback; assine novamente em um 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);
        }
    }
}

Pontos-chave: o esquema ssl:// ativa o TLS; setAutomaticReconnect(true) trata as quedas; setCleanStart(false) com uma expiração de sessão mantém as assinaturas durante interrupções curtas. Assinar novamente em connectComplete cobre as mais longas, mas isso roda em uma thread worker, porque chamar métodos bloqueantes do cliente a partir de um callback do Paho pode travar o cliente.

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

O builder se lê quase como uma frase. sslWithDefaultConfig() ativa o TLS com o trust store do JDK, e automaticReconnectWithDefaultConfig() reconecta com back-off exponencial. As propriedades do MQTT 5 contentType e messageExpiryInterval informam aos consumidores o formato do payload e permitem que o broker descarte leituras antigas após um minuto. A visão geral dos recursos do MQTT 5 explica essas propriedades em mais detalhes.

Bloqueante, assíncrono ou reativo?

Os dois exemplos usam um loop while simples com Thread.sleep para que o fluxo seja fácil de acompanhar, mas serviços reais raramente são assim. O estilo de API escolhido molda o resto do código, então vale decidir cedo.

Chamadas bloqueantes são as mais fáceis de ler: connect, subscribe e publish só retornam quando o broker responde. O MqttClient do Paho e o Mqtt5BlockingClient do HiveMQ funcionam assim. Eles combinam com ferramentas de linha de comando, testes e gateways pequenos, onde uma thread por conexão é aceitável.

Chamadas assíncronas retornam imediatamente com um token (IMqttToken no MqttAsyncClient do Paho) ou um CompletableFuture (HiveMQ). Elas combinam com serviços de backend que publicam em altas taxas, porque uma confirmação lenta nunca trava a thread produtora. Encadeie o tratamento de erros no future em vez de ignorá-lo, como o exemplo do HiveMQ faz com whenComplete.

Streams reativos, disponíveis apenas no cliente HiveMQ por meio da API RxJava, modelam as mensagens recebidas como um Flowable com back-pressure real. Se a sua aplicação já usa Reactor ou RxJava, isso evita acumular mensagens em memória quando os consumidores ficam para trás.

Seja qual for a escolha, mantenha o processamento de mensagens fora da thread de rede. Um insert no banco de dados ou uma chamada HTTP dentro de messageArrived ou de um callback do HiveMQ atrasa todas as outras mensagens daquela conexão. Coloque o payload em uma fila ou executor e retorne rapidamente.

Design de topics para serviços Java

Os exemplos usam três famílias de topics: .../telemetry para leituras, .../cmd/# para comandos e .../status para disponibilidade. Manter comandos e dados separados permite que um consumidor assine a telemetria com um único curinga sem nunca ver comandos, e as permissões podem depois ser pensadas por família de topics. Coloque nos níveis do topic os identificadores pelos quais você vai filtrar, como site, linha e dispositivo, em vez de colocá-los apenas no payload JSON, porque o broker só consegue rotear por topics.

Testando os dois clientes

Execute qualquer uma das classes e acompanhe os topics a partir de um 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'

Rode apenas um cliente por vez com os IDs padrão ou dê a cada um seu próprio ID; dois clientes não podem compartilhar uma sessão.

Migrando para um broker privado

Para produção, crie um broker no portal da ConnectMQ e use o usuário, a senha e o client ID gerados. O client ID é vinculado às credenciais emitidas, então precisa corresponder exatamente:

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

O Paho lê a URI completa de MQTT_URL (ssl://broker.connectmq.com:8883); o exemplo do HiveMQ lê MQTT_HOST.

Checklist do cliente MQTT Java

1. Client ID único por JVM. Várias instâncias com o mesmo ID derrubam umas às outras.
2. QoS 1 por padrão. O plano Free suporta QoS 0 e 1; o QoS 2 exige um plano pago. Veja níveis de QoS do MQTT explicados.
3. Não bloqueie os callbacks. Passe o trabalho para um executor nas duas bibliotecas.
4. Trate falhas de publicação. Decida se vai descartar, armazenar em buffer ou tentar de novo enquanto estiver desconectado.
5. Registre os reason codes. Desconexões do MQTT 5 trazem um motivo, por exemplo 0x97 quando uma cota é excedida.
6. Mantenha as credenciais fora do código, em variáveis de ambiente ou em um cofre de segredos.

Solução de problemas

A página oficial do projeto Paho está em eclipse.dev/paho.

Mais tutoriais de backend

Se a sua equipe também escreve em Go ou Rust, compare com o tutorial de cliente MQTT em Go. Para um panorama mais amplo entre linguagens, leia o guia de bibliotecas de cliente MQTT.

Conecte seus serviços Java à ConnectMQ

Um broker MQTT gerenciado permite que seus serviços Java se conectem a um único endpoint TLS sem nenhum broker para operar. A ConnectMQ suporta MQTT 3.1.1 e 5.0, então tanto o Paho quanto o cliente HiveMQ funcionam sem alterações, e o explorador de topics em tempo real do portal mostra exatamente o que o seu código publica. Se você ainda está comparando opções de hospedagem, o guia de broker MQTT na nuvem aborda os trade-offs. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito.

Perguntas frequentes

Qual é a melhor biblioteca de cliente MQTT para Java?

As duas são sólidas. O Paho não tem dependências e é amplamente documentado; o HiveMQ MQTT Client tem uma API fluente moderna, suporte assíncrono e reativo e um desenvolvimento mais ativo.

O Eclipse Paho suporta MQTT 5 em Java?

Sim, por meio do artefato separado org.eclipse.paho.mqttv5.client. O artefato mais antigo, org.eclipse.paho.client.mqttv3, suporta apenas MQTT 3.1.1.

Como ativo TLS em um cliente MQTT Java?

Com o Paho, use uma URI ssl://host:8883. Com o cliente HiveMQ, chame sslWithDefaultConfig() e use a porta 8883. Ambos dependem do trust store do JDK para certificados de confiança pública.

Os clientes MQTT Java reconectam automaticamente?

Sim, se ativado: setAutomaticReconnect(true) no Paho e automaticReconnectWithDefaultConfig() no cliente HiveMQ. Assine novamente após as reconexões quando a sessão puder ter expirado.