Spring Boot MQTT Integration: Gerätedaten konsumieren und per REST bereitstellen

Ein Spring-Boot-MQTT-Service ist ein gängiger Weg, eine Geräteflotte an die übrigen Geschäftssysteme anzubinden: Die Geräte veröffentlichen über MQTT, während Ihre Web-Apps, Partner oder ein MES/ERP eine vertraute REST-API nutzen. In der Fertigung ist das ein typisches Muster der OT/IT-Integration: Maschinendaten kommen per MQTT aus der Halle, Java-Backends stellen sie den Geschäftssystemen bereit. In diesem Tutorial verwenden Sie die MQTT-5-Adapter von Spring Integration, um Telemetrie über TLS zu empfangen, den letzten Messwert pro Gerät zu speichern, ihn unter /api/devices bereitzustellen und Befehle an Geräte zurückzusenden – alles über eine einzige geteilte MQTT-Verbindung.

Architektur

Der Service steht zwischen zwei Welten:

Das für den Produktivbetrieb wichtigste Detail ist die Verbindung. Erstellen Inbound-Adapter und Outbound-Handler jeweils einen eigenen MQTT-Client, brauchen sie zwei Client IDs und zwei Verbindungen. Viele verwaltete MQTT-Broker, darunter ConnectMQ, vergeben eine an einen Satz Zugangsdaten gebundene Client ID; zwei Verbindungen mit derselben ID würden sich daher ständig gegenseitig trennen. Spring Integration 6 löst das mit einem ClientManager, den sich beide Komponenten teilen. Dieses Tutorial nutzt ihn von Anfang an.

Projekt-Setup

Erstellen Sie ein Spring-Boot-3-Projekt (Java 17+) mit diesen Abhängigkeiten in der 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>

Der Paho-MQTT-5-Client ist eine optionale Abhängigkeit von spring-integration-mqtt und muss daher explizit hinzugefügt werden. Wird er vergessen, ist das der häufigste Grund für ClassNotFoundException: org.eclipse.paho.mqttv5... beim Start.

Konfiguration

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}

Die Standardwerte zeigen auf den kostenlosen öffentlichen MQTT-Broker unter broker.connectmq.com, der anonyme Clients mit jeder nicht leeren, eindeutigen Client ID akzeptiert. Topics auf dem öffentlichen MQTT-Broker sind für alle sichtbar, er ist für Tests und Prototypen gedacht; verwenden Sie daher ein zufälliges Präfix und nur Testdaten, keine Produktionsdaten. Das Schema ssl:// mit Port 8883 aktiviert TLS; dem Let's-Encrypt-Zertifikat des Brokers vertraut das JDK von Haus aus.

Die MQTT-Konfigurationsklasse

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

Adapter und Handler erhalten denselben Mqttv5ClientManager, sodass die Anwendung genau eine MQTT-Verbindung mit einer Client ID öffnet. Der automatische Reconnect ist in den Verbindungsoptionen konfiguriert; mit setCleanStart(false) und einem Session-Expiry-Intervall behält der Broker die Subscription über kurze Netzwerkabbrüche hinweg.

Das Outbound-Gateway

Ein Messaging-Gateway bietet dem Rest der Anwendung ein einfaches Java-Interface zum Veröffentlichen:

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

Der v5-Handler wandelt einen String-Payload in UTF-8-Bytes um, und der Header MqttHeaders.TOPIC legt das Ziel-Topic pro Nachricht fest.

Die neuesten Messwerte speichern

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

Der Inbound-Adapter liefert standardmäßig byte[]-Payloads, und das empfangene Topic steht im Header MqttHeaders.RECEIVED_TOPIC. Die Geräte-ID ist das Segment direkt vor telemetry.

Der REST-Controller

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

Die ID-Prüfung ist keine Zierde. Ohne sie könnte ein Aufrufer eine ID mit /, + oder # senden und auf Topics veröffentlichen, die Sie nie freigeben wollten.

Wie eine Nachricht durch die Anwendung fließt

Es hilft, eine Telemetrienachricht von Anfang bis Ende zu verfolgen, denn Spring Integration verbirgt viel Infrastruktur hinter wenigen Beans.

1. Das Gerät veröffentlicht {"temperature":22.5} auf <prefix>/devices/sensor-1/telemetry.
2. Der Broker leitet die Nachricht an jede passende Subscription weiter, auch an die, die der Inbound-Adapter über den geteilten Client Manager angelegt hat.
3. Der Adapter verpackt die Payload-Bytes in eine Spring-Message und fügt Header wie RECEIVED_TOPIC, RECEIVED_QOS und RECEIVED_RETAINED hinzu.
4. Der telemetryFlow übergibt die Nachricht an ReadingStore.accept, das das JSON parst und den vorherigen Messwert dieses Geräts ersetzt.
5. Ein REST-Aufruf an /api/devices/sensor-1 liest die Map und gibt den neuesten Messwert als JSON zurück.

Befehle nehmen den umgekehrten Weg: Der Controller ruft das Gateway auf, das Gateway sendet eine Nachricht an den Kanal mqttOutbound, der Service Activator übergibt sie an den Outbound-Handler, und dieser veröffentlicht über dieselbe geteilte Verbindung. Da der Outbound-Handler asynchron arbeitet, liefert der REST-Aufruf 202 Accepted, ohne auf die Bestätigung des Brokers zu warten. Muss Ihre API die Zustellung bestätigen, schalten Sie den Handler in den synchronen Modus und nehmen die zusätzliche Latenz in Kauf.

Fehlerbehandlung

Standardmäßig wird eine Exception innerhalb des Flows protokolliert und die Nachricht verworfen. Für alles, was über einen Prototyp hinausgeht, fügen Sie dem Inbound-Adapter mit adapter.setErrorChannelName("mqttErrors") einen Error Channel hinzu und dazu einen kleinen Flow, der Topic und Payload protokolliert – so wird eine fehlerhafte Nachricht eines Geräts sichtbar, statt stillschweigend ignoriert zu werden. Halten Sie den Handler schnell: Wenn Sie Datenbankschreibvorgänge ergänzen, erwägen Sie einen Executor Channel zwischen Adapter und Store, damit eine langsame Datenbank die Nachrichtenzustellung nicht aufhält.

End-to-End-Test

Starten Sie die Anwendung mit ./mvnw spring-boot:run und simulieren Sie dann ein Gerät:

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

Fragen Sie die REST-API ab:

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

Senden Sie einen Befehl und beobachten Sie, wie er auf dem Geräte-Topic ankommt:

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

Hinweise für Spring Boot MQTT im Produktivbetrieb

1. Ausgestellte Zugangsdaten verwenden. Erstellen Sie einen Broker im ConnectMQ-Portal und setzen Sie MQTT_CLIENT_ID, MQTT_USERNAME und MQTT_PASSWORD auf die generierten Werte; die Client ID muss exakt übereinstimmen.
2. Eine Instanz, eine Client ID. Skalieren Sie auf mehrere Replikas, braucht jede eigene Zugangsdaten. Für lastverteilten Konsum derselben Telemetrie stehen MQTT-5-Shared-Subscriptions in den kostenpflichtigen Serverless-Plänen zur Verfügung.
3. Wichtiges persistieren. Die In-Memory-Map geht beim Neustart verloren. Für eine Historie schreiben Sie Messwerte im Flow in eine Datenbank oder lassen ein eigenes Ingest-Tool parallel zu diesem Service abonnieren.
4. QoS bewusst wählen. QoS 1 ist der richtige Standard für Telemetrie und Befehle; siehe MQTT-QoS-Stufen erklärt. Der Free-Plan unterstützt QoS bis 1.
5. Health bereitstellen. Ergänzen Sie Spring Boot Actuator und alarmieren Sie, wenn die MQTT-Verbindung länger als erwartet unterbrochen ist.

Fehlerbehebung

Die vollständige Adapter-Referenz finden Sie in der Spring-Integration-MQTT-Dokumentation.

Verwandte Leitfäden

Brauchen Sie einen reinen Java-Client ohne Spring, vergleicht das Java-MQTT-Client-Tutorial Paho mit dem HiveMQ-Client. Das Go-MQTT-Client-Tutorial baut dasselbe Muster in Go, und der Leitfaden zu MQTT-Client-Bibliotheken behandelt weitere Ökosysteme.

Spring Boot mit ConnectMQ verbinden

Ein verwalteter MQTT-Broker gibt Ihrem Spring-Boot-Service einen stabilen, verschlüsselten Endpunkt, während Sie sich auf die API konzentrieren. ConnectMQ unterstützt MQTT 5 mit Session Expiry und in kostenpflichtigen Plänen auch Shared Subscriptions; die Geräteansicht und der Live-Topic-Explorer helfen Ihnen zu prüfen, was Ihre Flotte sendet. Einen breiteren Blick auf die Hosting-Optionen bietet der Leitfaden zum Cloud-MQTT-Broker. Erstellen Sie unter app.connectmq.com ohne Kreditkarte einen kostenlosen Serverless-Broker und setzen Sie die oben genannten Umgebungsvariablen.

Häufig gestellte Fragen

Wie nutze ich MQTT in Spring Boot?

Fügen Sie spring-integration-mqtt und den Paho-MQTT-5-Client hinzu und konfigurieren Sie dann einen Mqttv5PahoMessageDrivenChannelAdapter für eingehende und einen Mqttv5PahoMessageHandler für ausgehende Nachrichten.

Können sich Inbound- und Outbound-MQTT-Adapter eine Verbindung teilen?

Ja. Seit Spring Integration 6 übergeben Sie beiden denselben Mqttv5ClientManager, der eine einzige Client ID und Verbindung nutzt.

Unterstützt Spring Integration MQTT 5?

Ja, über die Adapter mit dem Präfix Mqttv5, die auf dem Eclipse-Paho-MQTT-5-Client aufbauen.

Wie verbinde ich Spring Boot über TLS mit einem MQTT-Broker?

Verwenden Sie in den Verbindungsoptionen eine Server-URI der Form ssl://host:8883. Broker mit öffentlich vertrauenswürdigen Zertifikaten funktionieren mit dem Standard-Truststore des JDK.