Integração MQTT com Spring Boot: Consuma Dados de Dispositivos e Exponha-os via REST

Um serviço Spring Boot MQTT é uma forma comum de conectar uma frota de dispositivos, seja de uma mineradora, de uma usina ou de uma planta de manufatura, ao restante dos sistemas da empresa: os dispositivos publicam via MQTT, e as suas aplicações web e parceiros usam uma API REST familiar. Neste tutorial você vai usar os adaptadores MQTT 5 do Spring Integration para consumir telemetria via TLS, manter a leitura mais recente de cada dispositivo, expô-la em /api/devices e enviar comandos de volta aos dispositivos, tudo por meio de uma única conexão MQTT compartilhada.

Arquitetura

O serviço fica entre dois mundos:

O detalhe que mais importa em produção é a conexão. Se o adaptador de entrada e o handler de saída criarem cada um o próprio cliente MQTT, eles precisarão de dois client IDs e duas conexões. Muitos brokers MQTT gerenciados, incluindo a ConnectMQ, emitem um client ID vinculado a um conjunto de credenciais, então duas conexões com o mesmo ID ficariam desconectando uma à outra continuamente. O Spring Integration 6 resolve isso com um ClientManager compartilhado pelos dois componentes. Este tutorial o usa desde o início.

Configuração do projeto

Crie um projeto Spring Boot 3 (Java 17+) com estas dependências no pom.xml:

<parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>3.3.5</version>
</parent>

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-integration</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
  </dependency>
  <dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.mqttv5.client</artifactId>
    <version>1.2.5</version>
  </dependency>
</dependencies>

O cliente Paho MQTT 5 é uma dependência opcional do spring-integration-mqtt, então você precisa adicioná-lo explicitamente. Esquecê-lo é o motivo mais comum de ClassNotFoundException: org.eclipse.paho.mqttv5... na inicialização.

Configuração

src/main/resources/application.yml:

mqtt:
  url: ${MQTT_URL:ssl://broker.connectmq.com:8883}
  client-id: ${MQTT_CLIENT_ID:spring-s8d2lf}
  username: ${MQTT_USERNAME:}
  password: ${MQTT_PASSWORD:}
  prefix: ${MQTT_PREFIX:connectmq/demo/s8d2lf}

Os valores padrão apontam para o servidor MQTT público (gratuito) em broker.connectmq.com, que aceita clientes anônimos com qualquer client ID não vazio e único. Os topics do servidor público são visíveis para todos, então use um prefixo aleatório e apenas dados de teste. O esquema ssl:// com a porta 8883 ativa o TLS; o certificado Let's Encrypt do broker é confiável para o JDK sem configuração adicional.

A classe de configuração MQTT

package com.example.mqttdemo;

import java.nio.charset.StandardCharsets;

import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.mqtt.core.Mqttv5ClientManager;
import org.springframework.integration.mqtt.inbound.Mqttv5PahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.Mqttv5PahoMessageHandler;
import org.springframework.messaging.MessageHandler;

@Configuration
@IntegrationComponentScan
public class MqttConfig {

    @Bean
    public Mqttv5ClientManager mqttClientManager(
            @Value("${mqtt.url}") String url,
            @Value("${mqtt.client-id}") String clientId,
            @Value("${mqtt.username}") String username,
            @Value("${mqtt.password}") String password) {
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setServerURIs(new String[] { url });
        options.setAutomaticReconnect(true);
        options.setCleanStart(false);
        options.setSessionExpiryInterval(600L);
        options.setKeepAliveInterval(30);
        if (!username.isBlank()) {
            options.setUserName(username);
            options.setPassword(password.getBytes(StandardCharsets.UTF_8));
        }
        return new Mqttv5ClientManager(options, clientId);
    }

    @Bean
    public Mqttv5PahoMessageDrivenChannelAdapter mqttInbound(
            Mqttv5ClientManager clientManager,
            @Value("${mqtt.prefix}") String prefix) {
        Mqttv5PahoMessageDrivenChannelAdapter adapter =
                new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, prefix + "/devices/+/telemetry");
        adapter.setQos(1);
        return adapter;
    }

    @Bean
    public IntegrationFlow telemetryFlow(Mqttv5PahoMessageDrivenChannelAdapter mqttInbound,
                                         ReadingStore store) {
        return IntegrationFlow.from(mqttInbound)
                .handle(message -> store.accept(message))
                .get();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutbound")
    public MessageHandler mqttOutboundHandler(Mqttv5ClientManager clientManager) {
        Mqttv5PahoMessageHandler handler = new Mqttv5PahoMessageHandler(clientManager);
        handler.setAsync(true);
        handler.setDefaultQos(1);
        return handler;
    }
}

O adaptador e o handler recebem o mesmo Mqttv5ClientManager, então a aplicação abre exatamente uma conexão MQTT com um único client ID. A reconexão automática fica nas opções de conexão; com setCleanStart(false) e um intervalo de expiração de sessão, o broker mantém a assinatura durante quedas curtas de rede.

O gateway de saída

Um messaging gateway dá ao restante da aplicação uma interface Java simples para publicar:

package com.example.mqttdemo;

import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;

@MessagingGateway(defaultRequestChannel = "mqttOutbound")
public interface MqttGateway {
    void publish(@Header(MqttHeaders.TOPIC) String topic, String payload);
}

O handler v5 converte um payload String em bytes UTF-8, e o header MqttHeaders.TOPIC define o topic de destino por mensagem.

Armazenando as leituras mais recentes

package com.example.mqttdemo;

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Component
public class ReadingStore {

    public record DeviceReading(String deviceId, JsonNode data, Instant receivedAt) { }

    private static final Logger log = LoggerFactory.getLogger(ReadingStore.class);
    private final Map<String, DeviceReading> latest = new ConcurrentHashMap<>();
    private final ObjectMapper mapper;

    public ReadingStore(ObjectMapper mapper) {
        this.mapper = mapper;
    }

    public void accept(Message<?> message) {
        String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class);
        String[] parts = topic.split("/");
        String deviceId = parts[parts.length - 2];
        String json = new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);
        try {
            latest.put(deviceId, new DeviceReading(deviceId, mapper.readTree(json), Instant.now()));
        } catch (JsonProcessingException e) {
            log.warn("Ignoring invalid JSON from {}: {}", deviceId, e.getOriginalMessage());
        }
    }

    public Collection<DeviceReading> all() {
        return latest.values();
    }

    public Optional<DeviceReading> get(String deviceId) {
        return Optional.ofNullable(latest.get(deviceId));
    }
}

Por padrão, o adaptador de entrada entrega payloads byte[], e o topic recebido chega no header MqttHeaders.RECEIVED_TOPIC. O ID do dispositivo é o segmento imediatamente anterior a telemetry.

O controller REST

package com.example.mqttdemo;

import java.util.Collection;
import java.util.regex.Pattern;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/devices")
public class DeviceController {

    private static final Pattern SAFE_ID = Pattern.compile("[A-Za-z0-9_-]{1,64}");

    private final ReadingStore store;
    private final MqttGateway gateway;
    private final String prefix;

    public DeviceController(ReadingStore store, MqttGateway gateway,
                            @Value("${mqtt.prefix}") String prefix) {
        this.store = store;
        this.gateway = gateway;
        this.prefix = prefix;
    }

    @GetMapping
    public Collection<ReadingStore.DeviceReading> all() {
        return store.all();
    }

    @GetMapping("/{id}")
    public ResponseEntity<ReadingStore.DeviceReading> one(@PathVariable String id) {
        return ResponseEntity.of(store.get(id));
    }

    @PostMapping("/{id}/commands")
    public ResponseEntity<Void> command(@PathVariable String id, @RequestBody String body) {
        if (!SAFE_ID.matcher(id).matches()) {
            return ResponseEntity.badRequest().build();
        }
        gateway.publish(prefix + "/devices/" + id + "/cmd", body);
        return ResponseEntity.accepted().build();
    }
}

A verificação do ID não é enfeite. Sem ela, quem chama a API poderia enviar um ID contendo /, + ou # e publicar em topics que você nunca pretendeu expor.

Como uma mensagem percorre a aplicação

Vale acompanhar uma mensagem de telemetria do início ao fim, porque o Spring Integration esconde muita infraestrutura por trás de poucos beans.

1. O dispositivo publica {"temperature":22.5} em <prefix>/devices/sensor-1/telemetry.
2. O broker a roteia para todas as assinaturas correspondentes, incluindo a criada pelo adaptador de entrada por meio do client manager compartilhado.
3. O adaptador envolve os bytes do payload em uma Message do Spring e adiciona headers como RECEIVED_TOPIC, RECEIVED_QOS e RECEIVED_RETAINED.
4. O telemetryFlow passa a mensagem para ReadingStore.accept, que faz o parse do JSON e substitui a leitura anterior daquele dispositivo.
5. Uma chamada REST a /api/devices/sensor-1 lê o mapa e retorna a leitura mais recente como JSON.

Os comandos fazem o caminho inverso: o controller chama o gateway, o gateway envia uma mensagem ao channel mqttOutbound, o service activator a entrega ao handler de saída, e o handler publica pela mesma conexão compartilhada. Como o handler de saída é assíncrono, a chamada REST retorna 202 Accepted sem esperar a confirmação do broker. Se a sua API precisar confirmar a entrega, mude o handler para o modo síncrono e aceite a latência extra.

Tratamento de erros

Por padrão, uma exceção dentro do fluxo é registrada em log e a mensagem é descartada. Para qualquer coisa além de um protótipo, adicione um error channel ao adaptador de entrada com adapter.setErrorChannelName("mqttErrors") e um pequeno fluxo que registre o topic e o payload, para que uma mensagem malformada de um dispositivo fique visível em vez de ser ignorada silenciosamente. Mantenha o handler rápido: se você adicionar gravações em banco de dados, considere um executor channel entre o adaptador e o store para que um banco lento não segure a entrega de mensagens.

Testando de ponta a ponta

Inicie a aplicação com ./mvnw spring-boot:run e simule um dispositivo:

mosquitto_pub -h broker.connectmq.com -p 1883 -i dev-s8d2lf \
  -t 'connectmq/demo/s8d2lf/devices/sensor-1/telemetry' -m '{"temperature":22.5,"humidity":48}'

Consulte a API REST:

curl -s http://localhost:8080/api/devices
curl -s http://localhost:8080/api/devices/sensor-1

Envie um comando e veja-o chegar ao topic do dispositivo:

mosquitto_sub -h broker.connectmq.com -p 1883 -i cmdwatch-s8d2lf \
  -t 'connectmq/demo/s8d2lf/devices/+/cmd' -v &
curl -s -X POST http://localhost:8080/api/devices/sensor-1/commands \
  -H 'Content-Type: application/json' -d '{"fan":"on"}'

Notas de produção para Spring Boot MQTT

1. Use as credenciais emitidas. Crie um broker no portal da ConnectMQ e defina MQTT_CLIENT_ID, MQTT_USERNAME e MQTT_PASSWORD com os valores gerados; o client ID precisa corresponder exatamente.
2. Uma instância, um client ID. Se você escalar para várias réplicas, cada uma precisa de credenciais próprias. Para consumo com balanceamento de carga da mesma telemetria, as shared subscriptions do MQTT 5 estão disponíveis nos planos Serverless pagos.
3. Persista o que importa. O mapa em memória se perde ao reiniciar. Para manter histórico, grave as leituras em um banco de dados dentro do fluxo ou assine uma ferramenta de ingestão dedicada ao lado deste serviço.
4. Escolha o QoS de forma consciente. O QoS 1 é o padrão certo para telemetria e comandos; veja níveis de QoS do MQTT explicados. O plano Free suporta QoS até 1.
5. Exponha a saúde do serviço. Adicione o Spring Boot Actuator e dispare alertas quando a conexão MQTT ficar fora por mais tempo que o esperado.

Solução de problemas

A referência completa dos adaptadores está na documentação de MQTT do Spring Integration.

Guias relacionados

Se você precisa de um cliente Java puro, sem Spring, o tutorial de cliente MQTT em Java compara o Paho com o cliente HiveMQ. O tutorial de cliente MQTT em Go constrói o mesmo padrão em Go, e o guia de bibliotecas de cliente MQTT aborda outros ecossistemas.

Conecte o Spring Boot à ConnectMQ

Um broker MQTT gerenciado dá ao seu serviço Spring Boot um endpoint estável e criptografado enquanto você se concentra na API. A ConnectMQ suporta MQTT 5 com expiração de sessão e shared subscriptions nos planos pagos, e a visão de dispositivos e o explorador de topics em tempo real ajudam a conferir o que a sua frota está enviando. Para uma visão mais ampla das opções de hospedagem, veja o guia de broker MQTT na nuvem. Crie um broker Serverless gratuito em app.connectmq.com sem cartão de crédito e defina as variáveis de ambiente acima.

Perguntas frequentes

Como uso MQTT no Spring Boot?

Adicione o spring-integration-mqtt e o cliente Paho MQTT 5 e depois configure um Mqttv5PahoMessageDrivenChannelAdapter para as mensagens de entrada e um Mqttv5PahoMessageHandler para as de saída.

Os adaptadores MQTT de entrada e de saída podem compartilhar uma conexão?

Sim. Desde o Spring Integration 6, passe o mesmo Mqttv5ClientManager para os dois, o que usa um único client ID e uma única conexão.

O Spring Integration suporta MQTT 5?

Sim, por meio dos adaptadores com prefixo Mqttv5, que são construídos sobre o cliente Eclipse Paho MQTT 5.

Como conecto o Spring Boot a um broker MQTT via TLS?

Use uma URI de servidor ssl://host:8883 nas opções de conexão. Brokers com certificados de confiança pública funcionam com o trust store padrão do JDK.