MQTT mit Spring Boot und Spring Integration

Spring Boot hat keinen eigenen MQTT-Client, Spring Integration aber schon: Das Modul spring-integration-mqtt kapselt Eclipse Paho in Channel-Adaptern. Eingehende MQTT-Nachrichten werden so zu Spring-Message-Objekten auf einem Channel, und alles, was du an einen Outbound-Channel sendest, wird an den Broker gepublisht. In diesem Tutorial baust du einen kleinen Service, der abonniert, über ein Gateway-Interface publisht und sich Ende-zu-Ende aus dem Browser testen lässt.

Abhängigkeiten

Das Dependency Management von Spring Boot deckt die Spring-Integration-Version ab, du musst sie also nicht selbst festlegen:

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

Das Modul bringt den Paho-Client mqttv3 für MQTT 3.1.1 mit. Wenn du die MQTT-5-Adapter willst, füge zusätzlich org.eclipse.paho:org.eclipse.paho.mqttv5.client hinzu. Die Beispiele nutzen Spring Boot 3 oder neuer und Java 17.

Spring Integration ist die richtige Wahl, wenn MQTT einer von mehreren Integrationspunkten ist und du Channels, Transformer und Fehlerbehandlung in einem Modell haben willst. Braucht dein Service nur einen Client und willst du Reactive Streams oder volle MQTT-5-Kontrolle, ist der HiveMQ MQTT Client in einer einfachen Spring-Bean die schlankere Alternative.

Konfigurations-Properties

Halte Broker-Einstellungen aus dem Code heraus. Ein kleiner Record, gebunden mit @ConfigurationProperties, hält sie typisiert:

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

Aktiviere ihn mit @EnableConfigurationProperties(MqttProperties.class) auf einer Konfigurationsklasse. Nutze ssl:// mit Port 8883 für TLS und tcp:// mit 1883 für unverschlüsselte Verbindungen in der lokalen Entwicklung.

Die Client Factory

Alle Adapter beziehen ihre Verbindungen aus einer MqttPahoClientFactory. Hier konfigurierst du TLS, Zugangsdaten, Keep Alive und Reconnect-Verhalten an einer Stelle:

@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 ist die normale Paho-Klasse, also stehen alle Paho-Optionen zur Verfügung — Last Will, eigene SSLSocketFactory, MQTT-Version. Details zu jeder Option findest du in MQTT in Java mit Eclipse Paho.

Inbound: abonnieren mit einem Message-Driven-Adapter

MqttPahoMessageDrivenChannelAdapter abonniert einen oder mehrere Topic-Filter und schiebt jede empfangene Nachricht auf einen Output-Channel. Ein @ServiceActivator verarbeitet sie dann:

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

Nützliche Header auf jeder eingehenden Nachricht:

Der Standard-Converter macht aus Payloads Strings. Rufe converter.setPayloadAsBytes(true) auf, wenn du Binärformate wie CBOR oder Protobuf empfängst. Topics kannst du zur Laufzeit mit adapter.addTopic(...) und adapter.removeTopic(...) hinzufügen oder entfernen.

Fehler und langsame Handler

Mit einem DirectChannel läuft dein Handler auf Pahos Callback-Thread. Das hat zwei Folgen, die du einplanen solltest:

• Eine Exception im Handler geht an den Error-Channel des Adapters, wenn du einen mit adapter.setErrorChannel(...) setzt. Ohne ihn wird die Exception geloggt, und je nach Adapterversion und Einstellungen kann Paho den Fehler als Verbindungsverlust behandeln. Fange Fehler immer bewusst ab und route sie — etwa in einen Channel, der fehlgeschlagene Payloads auf ein Dead-Letter-Topic schreibt.
• Ein langsamer Handler verzögert das Acknowledgement von QoS-1-Nachrichten und blockiert jede andere Nachricht auf dieser Verbindung. Wenn die Verarbeitung eine Datenbank oder einen HTTP-Aufruf umfasst, stelle mqttInputChannel auf einen ExecutorChannel mit begrenztem Thread-Pool um, damit der MQTT-Thread die Nachricht nur übergibt.

Beachte: Gibst du die Nachricht an einen anderen Thread ab, wird das PUBACK gesendet, bevor deine Verarbeitung fertig ist. Stürzt der Service dazwischen ab, ist die Nachricht weg. Wenn At-least-once-Verarbeitung wirklich zählt, nutze adapter.setManualAcks(true) und bestätige aus deinem Handler über den Header IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, sobald die Arbeit festgeschrieben ist.

Outbound: publishen über ein Gateway

MqttPahoMessageHandler publisht alles, was seinen Input-Channel erreicht. Kombiniere ihn mit einem @MessagingGateway, damit der Rest deines Codes ein einfaches Java-Interface aufruft:

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

Jetzt kann jede Bean MqttGateway injizieren und gateway.send("{\"id\":42}", "testmqtt/spring/out/orders") aufrufen. setAsync(true) lässt den Aufruf zurückkehren, ohne auf das PUBACK zu warten; mit setAsyncEvents(true) publiziert Spring ein MqttMessageDeliveredEvent, sobald der Broker die Zustellung bestätigt.

Die Java-DSL-Alternative

Wenn du Flows lieber magst als einzelne Beans, drückt die Java DSL dieselbe Inbound-Pipeline in wenigen Zeilen aus, inklusive eines Transformationsschritts:

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

Prüfe mit dem Topic Matcher, welche Topics ein Filter wie testmqtt/spring/sensors/+/temp trifft, bevor du ihn deployst.

Verbindungs-Events und MQTT 5

Spring Integration publiziert Application Events für den MQTT-Lifecycle. Auf sie zu hören ist der einfachste Weg, Verbindungsprobleme zu loggen oder zu alarmieren:

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

Für MQTT 5 tauschst du die Klassen: Mqttv5PahoMessageDrivenChannelAdapter und Mqttv5PahoMessageHandler erhalten statt einer Factory Paho-v5-MqttConnectionOptions und mappen MQTT-5-User-Properties von und zu Message-Headern. Channels, Gateway und Handler bleiben gleich.

Skalieren mit Shared Subscriptions

Mit einem normalen Abonnement erhält jede Instanz deines Service jede Nachricht — drei Pods bedeuten, dass jede Bestellung dreimal verarbeitet wird. MQTT-5-Shared-Subscriptions (die viele Broker als Erweiterung auch für MQTT-3.1.1-Verbindungen unterstützen) lösen das: Abonniere $share/orders/testmqtt/spring/in/#, und der Broker verteilt die Nachrichten auf alle Mitglieder der Gruppe orders. Nur der Topic-String in application.yml ändert sich. Die Zustellregeln erklärt MQTT Shared Subscriptions.

Ende-zu-Ende testen

1. Starte die Anwendung und achte im Log auf das Abonnement.
2. Öffne den Online-MQTT-Client, verbinde dich mit demselben Broker und publishe hello auf testmqtt/spring/in/demo. Dein Service sollte received testmqtt/spring/in/demo: hello loggen.
3. Abonniere im Browser testmqtt/spring/out/#, rufe einen Endpoint auf, der MqttGateway nutzt, und prüfe, ob die Nachricht mit QoS 1 erscheint.
4. Unterbrich die Broker-Verbindung (zum Beispiel, indem du kurz das Netzwerk abschaltest) und prüfe, dass ein MqttConnectionFailedEvent geloggt wird und der Adapter sich neu verbindet.

Traffic deines Spring-Boot-Service beobachten

Kommen keine Nachrichten an, sagt die Paho-Exception im Log meist, warum — Paho-Java-Fehler listet die häufigsten auf, und die Reason-Code-Tabelle entschlüsselt Ablehnungen des Brokers.

Spring Integration macht MQTT zu einem weiteren Channel in deiner Spring-Boot-Anwendung: Konfiguriere die Client Factory einmal, verbinde Inbound-Adapter mit Service Activators, publishe über ein Gateway und halte Client-IDs pro Verbindung eindeutig. Sobald der Browser-Test besteht, läuft dieselbe Konfiguration in Produktion — nur die Zugangsdaten ändern sich.

Häufig gestellte Fragen