Intégration MQTT avec Spring Boot : consommer les données des appareils et les exposer via REST

Un service Spring Boot MQTT est un moyen courant de relier une flotte d'appareils au reste de vos systèmes métier : les appareils publient en MQTT, tandis que vos applications web et vos partenaires utilisent une API REST familière. Dans ce tutoriel, vous utiliserez les adaptateurs MQTT 5 de Spring Integration pour consommer la télémétrie sur TLS, conserver le dernier relevé de chaque appareil, l'exposer sur /api/devices et renvoyer des commandes aux appareils, le tout via une seule connexion MQTT partagée.

Architecture

Le service fait le pont entre deux mondes :

Le point le plus important pour la production est la connexion. Si l'adaptateur entrant et le handler sortant créent chacun leur propre client MQTT, il leur faut deux client ID et deux connexions. Or de nombreux brokers MQTT managés, dont ConnectMQ, délivrent un client ID lié à un jeu d'identifiants : deux connexions avec le même ID se déconnecteraient donc sans cesse l'une l'autre. Spring Integration 6 résout ce problème avec un ClientManager partagé par les deux composants. Ce tutoriel l'utilise dès le départ.

Configuration du projet

Créez un projet Spring Boot 3 (Java 17+) avec ces dépendances dans 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>

Le client Paho MQTT 5 est une dépendance optionnelle de spring-integration-mqtt : vous devez donc l'ajouter explicitement. L'oublier est la cause la plus fréquente d'une ClassNotFoundException: org.eclipse.paho.mqttv5... au démarrage.

Configuration

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}

Les valeurs par défaut pointent vers le serveur MQTT public gratuit sur broker.connectmq.com, qui accepte les clients anonymes avec n'importe quel client ID unique et non vide. Les topics de l'espace public sont visibles par tous : utilisez donc un préfixe aléatoire et uniquement des données de test. Le schéma ssl:// avec le port 8883 active TLS ; le certificat Let's Encrypt du broker est reconnu d'emblée par le JDK.

La classe de configuration 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;
    }
}

L'adaptateur et le handler reçoivent tous deux le même Mqttv5ClientManager : l'application ouvre donc exactement une connexion MQTT avec un seul client ID. La reconnexion automatique est définie dans les options de connexion ; avec setCleanStart(false) et un intervalle d'expiration de session, le broker conserve l'abonnement lors de courtes coupures réseau.

La passerelle sortante

Une messaging gateway offre au reste de l'application une simple interface Java pour publier :

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

Le handler v5 convertit un payload String en octets UTF-8, et l'en-tête MqttHeaders.TOPIC définit le topic de destination de chaque message.

Stocker les derniers relevés

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

Par défaut, l'adaptateur entrant livre des payloads byte[], et le topic reçu arrive dans l'en-tête MqttHeaders.RECEIVED_TOPIC. L'ID de l'appareil correspond au segment situé juste avant telemetry.

Le contrôleur 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();
    }
}

La vérification de l'ID n'est pas décorative. Sans elle, un appelant pourrait envoyer un ID contenant /, + ou # et publier sur des topics que vous n'avez jamais voulu exposer.

Parcours d'un message dans l'application

Il est utile de suivre un message de télémétrie de bout en bout, car Spring Integration masque beaucoup de plomberie derrière quelques beans.

1. L'appareil publie {"temperature":22.5} sur <prefix>/devices/sensor-1/telemetry.
2. Le broker le route vers tous les abonnements correspondants, y compris celui créé par l'adaptateur entrant via le client manager partagé.
3. L'adaptateur encapsule les octets du payload dans un Message Spring et ajoute des en-têtes comme RECEIVED_TOPIC, RECEIVED_QOS et RECEIVED_RETAINED.
4. Le telemetryFlow transmet le message à ReadingStore.accept, qui analyse le JSON et remplace le relevé précédent de cet appareil.
5. Un appel REST à /api/devices/sensor-1 lit la map et renvoie le dernier relevé en JSON.

Les commandes suivent le chemin inverse : le contrôleur appelle la gateway, la gateway envoie un message sur le canal mqttOutbound, le service activator le transmet au handler sortant, et le handler publie via la même connexion partagée. Comme le handler sortant est asynchrone, l'appel REST renvoie 202 Accepted sans attendre l'acquittement du broker. Si votre API doit confirmer la livraison, passez le handler en mode synchrone et acceptez la latence supplémentaire.

Gestion des erreurs

Par défaut, une exception dans le flow est journalisée et le message est abandonné. Pour tout ce qui dépasse le prototype, ajoutez un canal d'erreur à l'adaptateur entrant avec adapter.setErrorChannelName("mqttErrors") et un petit flow qui journalise le topic et le payload, afin qu'un message malformé provenant d'un appareil soit visible au lieu d'être ignoré en silence. Gardez le handler rapide : si vous ajoutez des écritures en base de données, envisagez un executor channel entre l'adaptateur et le store pour qu'une base lente ne retarde pas la livraison des messages.

Tests de bout en bout

Démarrez l'application avec ./mvnw spring-boot:run, puis simulez un appareil :

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

Interrogez l'API REST :

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

Envoyez une commande et observez son arrivée sur le topic de l'appareil :

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"}'

Notes de production pour Spring Boot MQTT

1. Utilisez les identifiants délivrés. Créez un broker dans le portail ConnectMQ et renseignez MQTT_CLIENT_ID, MQTT_USERNAME et MQTT_PASSWORD avec les valeurs générées ; le client ID doit correspondre exactement.
2. Une instance, un client ID. Si vous passez à plusieurs réplicas, chacun a besoin de ses propres identifiants. Pour répartir la consommation d'une même télémétrie, les shared subscriptions MQTT 5 sont disponibles sur les plans Serverless payants.
3. Persistez ce qui compte. La map en mémoire est perdue au redémarrage. Pour l'historique, écrivez les relevés en base de données dans le flow, ou abonnez un outil d'ingestion dédié en parallèle de ce service.
4. Choisissez la QoS de façon réfléchie. QoS 1 est le bon choix par défaut pour la télémétrie et les commandes ; voir les niveaux de QoS MQTT expliqués. Le plan Free prend en charge la QoS jusqu'à 1.
5. Exposez l'état de santé. Ajoutez Spring Boot Actuator et déclenchez une alerte lorsque la connexion MQTT reste coupée plus longtemps que prévu.

Dépannage

La référence complète des adaptateurs se trouve dans la documentation Spring Integration MQTT.

Guides associés

Si vous avez besoin d'un client Java simple sans Spring, le tutoriel client MQTT Java compare Paho et le client HiveMQ. Le tutoriel client MQTT en Go construit le même modèle en Go, et le guide des bibliothèques client MQTT couvre d'autres écosystèmes.

Connectez Spring Boot à ConnectMQ

Un broker MQTT managé offre à votre service Spring Boot un endpoint stable et chiffré pendant que vous vous concentrez sur l'API. ConnectMQ prend en charge MQTT 5 avec expiration de session et shared subscriptions sur les plans payants, et sa vue des appareils ainsi que son explorateur de topics en direct vous aident à vérifier ce qu'envoie votre flotte. Pour un panorama plus large des options d'hébergement, consultez le guide du broker MQTT cloud. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, et définissez les variables d'environnement ci-dessus.

Questions fréquentes

Comment utiliser MQTT dans Spring Boot ?

Ajoutez spring-integration-mqtt et le client Paho MQTT 5, puis configurez un Mqttv5PahoMessageDrivenChannelAdapter pour les messages entrants et un Mqttv5PahoMessageHandler pour les messages sortants.

Les adaptateurs MQTT entrant et sortant peuvent-ils partager une connexion ?

Oui. Depuis Spring Integration 6, passez le même Mqttv5ClientManager aux deux : il utilise un seul client ID et une seule connexion.

Spring Integration prend-il en charge MQTT 5 ?

Oui, via les adaptateurs préfixés par Mqttv5, qui reposent sur le client Eclipse Paho MQTT 5.

Comment connecter Spring Boot à un broker MQTT sur TLS ?

Utilisez une URI de serveur ssl://host:8883 dans les options de connexion. Les brokers dotés de certificats reconnus publiquement fonctionnent avec le trust store par défaut du JDK.