MQTT com Spring Boot e Spring Integration

O Spring Boot não tem um cliente MQTT próprio, mas o Spring Integration tem: o módulo spring-integration-mqtt encapsula o Eclipse Paho em channel adapters, de modo que mensagens MQTT recebidas viram objetos Message do Spring em um canal, e tudo o que você envia para um canal de saída é publicado no broker. Neste tutorial você constrói um pequeno serviço que assina, publica por meio de uma interface gateway e pode ser testado de ponta a ponta pelo navegador.

Dependências

O gerenciamento de dependências do Spring Boot cobre a versão do Spring Integration, então você não precisa fixá-la:

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

O módulo traz o cliente Paho mqttv3 para MQTT 3.1.1. Se quiser os adapters MQTT 5, adicione também org.eclipse.paho:org.eclipse.paho.mqttv5.client. Os exemplos abaixo usam Spring Boot 3 ou superior e Java 17.

O Spring Integration é a escolha certa quando o MQTT é um entre vários pontos de integração e você quer canais, transformers e tratamento de erros em um único modelo. Se o seu serviço só precisa de um cliente e você quer reactive streams ou controle total do MQTT 5, o HiveMQ MQTT Client encapsulado em um bean Spring comum é uma alternativa mais leve.

Propriedades de configuração

Mantenha as configurações do broker fora do código. Um pequeno record ligado com @ConfigurationProperties as mantém tipadas:

mqtt:
  url: ssl://broker.connectmq.com:8883
  username: ${MQTT_USER}
  password: ${MQTT_PASS}
  client-id: orders-service-${random.uuid}
  inbound-topic: testmqtt/spring/in/#
  outbound-topic: testmqtt/spring/out

@ConfigurationProperties(prefix = "mqtt")
public record MqttProperties(String url, String username, String password,
                             String clientId, String inboundTopic, String outboundTopic) {}

Ative com @EnableConfigurationProperties(MqttProperties.class) em uma classe de configuração. Use ssl:// com a porta 8883 para TLS e tcp:// com a 1883 para conexões sem criptografia no desenvolvimento local.

A client factory

Todos os adapters obtêm suas conexões de uma MqttPahoClientFactory. É aqui que TLS, credenciais, keep alive e comportamento de reconexão são configurados uma única vez:

@Configuration
@EnableConfigurationProperties(MqttProperties.class)
@IntegrationComponentScan
public class MqttConfig {

    @Bean
    public MqttPahoClientFactory mqttClientFactory(MqttProperties props) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[] { props.url() });
        options.setUserName(props.username());
        options.setPassword(props.password().toCharArray());
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        options.setKeepAliveInterval(30);
        options.setConnectionTimeout(10);

        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(options);
        return factory;
    }
}

MqttConnectOptions é a classe normal do Paho, então todas as opções do Paho estão disponíveis — Last Will, SSLSocketFactory personalizado, versão do MQTT. Veja MQTT em Java com Eclipse Paho para os detalhes de cada opção.

Entrada: assinar com um message-driven adapter

O MqttPahoMessageDrivenChannelAdapter assina um ou mais filtros de topic e envia cada mensagem recebida para um canal de saída. Um @ServiceActivator então a processa:

@Bean
public MessageChannel mqttInputChannel() {
    return new DirectChannel();
}

@Bean
public MessageProducer inbound(MqttPahoClientFactory factory, MqttProperties props) {
    MqttPahoMessageDrivenChannelAdapter adapter =
            new MqttPahoMessageDrivenChannelAdapter(props.clientId() + "-in", factory, props.inboundTopic());
    adapter.setCompletionTimeout(5000);
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(1);
    adapter.setOutputChannel(mqttInputChannel());
    return adapter;
}

@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler inboundHandler() {
    return message -> {
        String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class);
        System.out.println("received " + topic + ": " + message.getPayload());
    };
}

Headers úteis em cada mensagem recebida:

O converter padrão transforma payloads em String. Chame converter.setPayloadAsBytes(true) se você recebe formatos binários como CBOR ou Protobuf. Também é possível adicionar ou remover topics em tempo de execução com adapter.addTopic(...) e adapter.removeTopic(...).

Erros e handlers lentos

Com um DirectChannel, seu handler roda na thread de callback do Paho. Isso tem duas consequências que vale planejar:

• Uma exceção no handler é enviada ao canal de erro do adapter, se você configurar um com adapter.setErrorChannel(...). Sem ele, a exceção é registrada no log e, dependendo da versão e das configurações do adapter, o Paho pode tratar a falha como conexão perdida. Sempre capture e roteie erros de forma deliberada — por exemplo para um canal que grava payloads com falha em um topic de dead letter.
• Um handler lento atrasa o acknowledgement das mensagens QoS 1 e bloqueia todas as outras mensagens daquela conexão. Se o processamento envolve um banco de dados ou uma chamada HTTP, troque mqttInputChannel por um ExecutorChannel com um pool de threads limitado, para que a thread MQTT apenas repasse a mensagem.

Atenção: repassar a mensagem para outra thread significa que o PUBACK é enviado antes de o processamento terminar. Se o serviço cair no meio do caminho, essa mensagem se perde. Quando o processamento at-least-once realmente importa, use adapter.setManualAcks(true) e confirme a partir do handler com o header IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK depois que o trabalho for gravado.

Saída: publicar por um gateway

O MqttPahoMessageHandler publica tudo o que chega ao seu canal de entrada. Combine-o com um @MessagingGateway para que o resto do código chame uma interface Java comum:

@Bean
public MessageChannel mqttOutboundChannel() {
    return new DirectChannel();
}

@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound(MqttPahoClientFactory factory, MqttProperties props) {
    MqttPahoMessageHandler handler = new MqttPahoMessageHandler(props.clientId() + "-out", factory);
    handler.setAsync(true);
    handler.setAsyncEvents(true);
    handler.setDefaultTopic(props.outboundTopic());
    handler.setDefaultQos(1);
    return handler;
}

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {

    void send(String payload);

    void send(String payload, @Header(MqttHeaders.TOPIC) String topic);

    void send(String payload, @Header(MqttHeaders.TOPIC) String topic,
              @Header(MqttHeaders.QOS) int qos, @Header(MqttHeaders.RETAINED) boolean retained);
}

Agora qualquer bean pode injetar MqttGateway e chamar gateway.send("{\"id\":42}", "testmqtt/spring/out/orders"). setAsync(true) faz o envio retornar sem esperar o PUBACK; com setAsyncEvents(true), o Spring publica um MqttMessageDeliveredEvent quando o broker confirma a entrega.

A alternativa com Java DSL

Se você prefere flows a beans separados, o Java DSL expressa o mesmo pipeline de entrada em poucas linhas, incluindo uma etapa de transformação:

@Bean
public IntegrationFlow temperatureFlow(MqttPahoClientFactory factory, MqttProperties props) {
    return IntegrationFlow
            .from(new MqttPahoMessageDrivenChannelAdapter(
                    props.clientId() + "-temp", factory, "testmqtt/spring/sensors/+/temp"))
            .<String, Double>transform(Double::valueOf)
            .filter((Double t) -> t > 30.0)
            .handle(m -> System.out.println("too hot: " + m.getPayload()))
            .get();
}

Verifique quais topics um filtro como testmqtt/spring/sensors/+/temp corresponde no topic matcher antes do deploy.

Eventos de conexão e MQTT 5

O Spring Integration publica eventos de aplicação para o ciclo de vida do MQTT. Escutá-los é a forma mais simples de registrar ou alertar sobre problemas de conexão:

@Component
public class MqttEvents {

    @EventListener
    public void onFailure(MqttConnectionFailedEvent event) {
        System.err.println("MQTT connection failed: " + event.getCause());
    }

    @EventListener
    public void onSubscribed(MqttSubscribedEvent event) {
        System.out.println(event.getMessage());
    }
}

Para MQTT 5, troque as classes: Mqttv5PahoMessageDrivenChannelAdapter e Mqttv5PahoMessageHandler recebem um MqttConnectionOptions do Paho v5 em vez de uma factory e mapeiam as user properties do MQTT 5 de e para headers da mensagem. Canais, gateway e handlers continuam iguais.

Escalando com shared subscriptions

Com uma assinatura normal, cada instância do serviço recebe todas as mensagens — três pods significam que cada pedido é processado três vezes. As shared subscriptions do MQTT 5 (que muitos brokers também suportam como extensão em conexões MQTT 3.1.1) resolvem isso: assine $share/orders/testmqtt/spring/in/# e o broker distribui as mensagens entre todos os membros do grupo orders. Só o topic no application.yml muda. Leia shared subscriptions no MQTT para as regras de entrega.

Teste de ponta a ponta

1. Inicie a aplicação e procure a assinatura no log.
2. Abra o cliente MQTT online, conecte-se ao mesmo broker e publique hello em testmqtt/spring/in/demo. Seu serviço deve registrar received testmqtt/spring/in/demo: hello.
3. Assine testmqtt/spring/out/# no navegador, chame um endpoint que use MqttGateway e confira se a mensagem aparece com QoS 1.
4. Interrompa a conexão com o broker (por exemplo, desativando a rede por um instante) e confirme que um MqttConnectionFailedEvent é registrado e que o adapter se reconecta.

Acompanhe o tráfego do seu serviço Spring Boot

Se as mensagens não chegarem, a exceção do Paho no log geralmente diz o motivo — erros do Paho Java lista os mais comuns, e a tabela de reason codes decodifica as recusas do broker.

O Spring Integration transforma o MQTT em apenas mais um canal da sua aplicação Spring Boot: configure a client factory uma vez, ligue os adapters de entrada a service activators, publique por um gateway e mantenha os client IDs únicos por conexão. Depois que o teste no navegador passar, a mesma configuração roda em produção trocando apenas as credenciais.

Perguntas frequentes