MQTT avec Spring Boot et Spring Integration

Spring Boot n'a pas de client MQTT propre, mais Spring Integration en a un : le module spring-integration-mqtt encapsule Eclipse Paho dans des channel adapters. Les messages MQTT entrants deviennent des objets Message Spring sur un canal, et tout ce que vous envoyez sur un canal sortant est publié sur le broker. Dans ce tutoriel, vous construisez un petit service qui s'abonne, publie via une interface gateway et se teste de bout en bout depuis le navigateur.

Dépendances

La gestion des dépendances de Spring Boot couvre la version de Spring Integration, vous n'avez donc pas à la fixer vous-même :

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

Le module apporte le client Paho mqttv3 pour MQTT 3.1.1. Si vous voulez les adapters MQTT 5, ajoutez aussi org.eclipse.paho:org.eclipse.paho.mqttv5.client. Les exemples ci-dessous utilisent Spring Boot 3 ou plus récent et Java 17.

Spring Integration est le bon choix quand MQTT est un point d'intégration parmi d'autres et que vous voulez canaux, transformateurs et gestion d'erreurs dans un même modèle. Si votre service n'a besoin que d'un client et que vous voulez des reactive streams ou un contrôle complet de MQTT 5, le HiveMQ MQTT Client encapsulé dans un simple bean Spring est une alternative plus légère.

Propriétés de configuration

Gardez les paramètres du broker hors du code. Un petit record lié avec @ConfigurationProperties les garde typés :

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) {}

Activez-le avec @EnableConfigurationProperties(MqttProperties.class) sur une classe de configuration. Utilisez ssl:// avec le port 8883 pour TLS et tcp:// avec le 1883 pour les connexions en clair pendant le développement local.

La client factory

Tous les adapters obtiennent leurs connexions d'une MqttPahoClientFactory. C'est là que TLS, identifiants, keep alive et comportement de reconnexion sont configurés une seule fois :

@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 est la classe Paho habituelle : toutes les options Paho sont donc disponibles — Last Will, SSLSocketFactory personnalisée, version MQTT. Consultez MQTT en Java avec Eclipse Paho pour le détail de chaque option.

Entrant : s'abonner avec un message-driven adapter

MqttPahoMessageDrivenChannelAdapter s'abonne à un ou plusieurs filtres de topic et pousse chaque message reçu vers un canal de sortie. Un @ServiceActivator le traite ensuite :

@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 utiles sur chaque message entrant :

Le converter par défaut transforme les payloads en String. Appelez converter.setPayloadAsBytes(true) si vous recevez des formats binaires comme CBOR ou Protobuf. Vous pouvez aussi ajouter ou retirer des topics à l'exécution avec adapter.addTopic(...) et adapter.removeTopic(...).

Erreurs et handlers lents

Avec un DirectChannel, votre handler s'exécute sur le thread de callback de Paho. Cela a deux conséquences à anticiper :

• Une exception dans le handler est envoyée au canal d'erreur de l'adapter si vous en définissez un avec adapter.setErrorChannel(...). Sinon, l'exception est journalisée et, selon la version et la configuration de l'adapter, Paho peut traiter l'échec comme une perte de connexion. Capturez et routez toujours les erreurs délibérément, par exemple vers un canal qui écrit les payloads en échec sur un topic de dead letter.
• Un handler lent retarde l'acquittement des messages QoS 1 et bloque tous les autres messages de cette connexion. Si le traitement implique une base de données ou un appel HTTP, remplacez mqttInputChannel par un ExecutorChannel adossé à un pool de threads borné, pour que le thread MQTT se contente de transmettre le message.

Attention : confier le message à un autre thread signifie que le PUBACK part avant la fin de votre traitement. Si le service plante entre les deux, ce message est perdu. Quand un traitement at-least-once compte vraiment, utilisez adapter.setManualAcks(true) et acquittez depuis votre handler via le header IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK une fois le travail validé.

Sortant : publier via un gateway

MqttPahoMessageHandler publie tout ce qui arrive sur son canal d'entrée. Associez-le à un @MessagingGateway pour que le reste de votre code appelle une simple interface Java :

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

Désormais, n'importe quel bean peut injecter MqttGateway et appeler gateway.send("{\"id\":42}", "testmqtt/spring/out/orders"). setAsync(true) fait revenir l'envoi sans attendre le PUBACK ; avec setAsyncEvents(true), Spring publie un MqttMessageDeliveredEvent quand le broker confirme la livraison.

L'alternative Java DSL

Si vous préférez les flows à des beans séparés, le Java DSL exprime le même pipeline entrant en quelques lignes, étape de transformation comprise :

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

Vérifiez quels topics un filtre comme testmqtt/spring/sensors/+/temp correspond grâce au topic matcher avant de le déployer.

Événements de connexion et MQTT 5

Spring Integration publie des événements applicatifs pour le cycle de vie MQTT. Les écouter est le moyen le plus simple de journaliser ou d'alerter sur les problèmes de connexion :

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

Pour MQTT 5, changez de classes : Mqttv5PahoMessageDrivenChannelAdapter et Mqttv5PahoMessageHandler prennent des MqttConnectionOptions Paho v5 au lieu d'une factory, et mappent les user properties MQTT 5 vers et depuis les headers du message. Les canaux, le gateway et les handlers ne changent pas.

Passer à l'échelle avec les shared subscriptions

Avec un abonnement classique, chaque instance de votre service reçoit chaque message : trois pods signifient que chaque commande est traitée trois fois. Les shared subscriptions de MQTT 5 (que de nombreux brokers prennent aussi en charge en extension sur les connexions MQTT 3.1.1) règlent ce problème : abonnez-vous à $share/orders/testmqtt/spring/in/# et le broker répartit les messages entre tous les membres du groupe orders. Seul le topic dans application.yml change. Lisez les shared subscriptions MQTT pour les règles de livraison.

Tester de bout en bout

1. Démarrez l'application et repérez l'abonnement dans les logs.
2. Ouvrez le client MQTT en ligne, connectez-vous au même broker et publiez hello sur testmqtt/spring/in/demo. Votre service doit journaliser received testmqtt/spring/in/demo: hello.
3. Abonnez-vous dans le navigateur à testmqtt/spring/out/#, appelez un endpoint qui utilise MqttGateway et vérifiez que le message apparaît en QoS 1.
4. Coupez la connexion au broker (par exemple en désactivant brièvement le réseau) et vérifiez qu'un MqttConnectionFailedEvent est journalisé et que l'adapter se reconnecte.

Observez le trafic de votre service Spring Boot

Si les messages n'arrivent pas, l'exception Paho dans les logs en donne généralement la raison : les erreurs Paho Java recense les plus courantes, et la table des reason codes décode les refus du broker.

Spring Integration fait de MQTT un canal de plus dans votre application Spring Boot : configurez la client factory une fois, reliez les adapters entrants à des service activators, publiez via un gateway et gardez des client IDs uniques par connexion. Une fois le test navigateur réussi, la même configuration tourne en production en changeant uniquement les identifiants.

Questions fréquentes