Spring Boot MQTT Integration: Consume Device Data and Expose It via REST

A Spring Boot MQTT service is a common way to connect a fleet of devices to the rest of your business systems: devices publish over MQTT, and your web apps and partners use a familiar REST API. In this tutorial you will use Spring Integration's MQTT 5 adapters to consume telemetry over TLS, keep the latest reading per device, expose it at /api/devices, and send commands back to devices, all through a single shared MQTT connection.

Architecture

The service sits between two worlds:

The detail that matters most for production is the connection. If the inbound adapter and outbound handler each create their own MQTT client, they need two client IDs and two connections. Many hosted brokers, including ConnectMQ, issue a client ID bound to a credential set, so two connections with one ID would continually disconnect each other. Spring Integration 6 solves this with a ClientManager that both components share. This tutorial uses it from the start.

Project Setup

Create a Spring Boot 3 project (Java 17+) with these dependencies in 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>

The Paho MQTT 5 client is an optional dependency of spring-integration-mqtt, so you must add it explicitly. Forgetting it is the most common reason for ClassNotFoundException: org.eclipse.paho.mqttv5... at startup.

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}

The defaults point at the free public MQTT broker at broker.connectmq.com, which accepts anonymous clients with any non-empty, unique client ID. Topics in the public area are visible to everyone, so use a random prefix and only test data. The ssl:// scheme with port 8883 enables TLS; the broker's Let's Encrypt certificate is trusted by the JDK out of the box.

The MQTT Configuration Class

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

Both the adapter and the handler receive the same Mqttv5ClientManager, so the application opens exactly one MQTT connection with one client ID. Automatic reconnect lives in the connection options; with setCleanStart(false) and a session expiry interval, the broker keeps the subscription across short network drops.

The Outbound Gateway

A messaging gateway gives the rest of the application a plain Java interface for publishing:

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

The v5 handler converts a String payload to UTF-8 bytes, and the MqttHeaders.TOPIC header sets the destination topic per message.

Storing the Latest Readings

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

The inbound adapter delivers byte[] payloads by default, and the received topic arrives in the MqttHeaders.RECEIVED_TOPIC header. The device ID is the segment just before telemetry.

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

The ID check is not decoration. Without it, a caller could send an ID containing /, + or # and publish to topics you never intended to expose.

How a Message Flows Through the Application

It helps to trace one telemetry message from start to finish, because Spring Integration hides a lot of plumbing behind a few beans.

1. The device publishes {"temperature":22.5} to <prefix>/devices/sensor-1/telemetry.
2. The broker routes it to every matching subscription, including the one created by the inbound adapter through the shared client manager.
3. The adapter wraps the payload bytes in a Spring Message and adds headers such as RECEIVED_TOPIC, RECEIVED_QOS and RECEIVED_RETAINED.
4. The telemetryFlow passes the message to ReadingStore.accept, which parses the JSON and replaces the previous reading for that device.
5. A REST call to /api/devices/sensor-1 reads the map and returns the latest reading as JSON.

Commands travel the opposite way: the controller calls the gateway, the gateway sends a message to the mqttOutbound channel, the service activator hands it to the outbound handler, and the handler publishes through the same shared connection. Because the outbound handler is asynchronous, the REST call returns 202 Accepted without waiting for the broker's acknowledgement. If your API must confirm delivery, switch the handler to synchronous mode and accept the extra latency.

Error handling

By default, an exception inside the flow is logged and the message is dropped. For anything beyond a prototype, add an error channel to the inbound adapter with adapter.setErrorChannelName("mqttErrors") and a small flow that logs the topic and payload, so a malformed message from one device is visible instead of silently ignored. Keep the handler fast: if you add database writes, consider an executor channel between the adapter and the store so a slow database does not hold up message delivery.

Testing End to End

Start the application with ./mvnw spring-boot:run, then simulate a device:

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

Query the REST API:

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

Send a command and watch it arrive on the device topic:

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

Production Notes for Spring Boot MQTT

1. Use issued credentials. Create a broker in the ConnectMQ portal and set MQTT_CLIENT_ID, MQTT_USERNAME and MQTT_PASSWORD to the generated values; the client ID must match exactly.
2. One instance, one client ID. If you scale to several replicas, each needs its own credentials. For load-balanced consumption of the same telemetry, MQTT 5 shared subscriptions are available on paid Serverless plans.
3. Persist what matters. The in-memory map is lost on restart. For history, write readings to a database inside the flow, or subscribe a dedicated ingest tool alongside this service.
4. Pick QoS deliberately. QoS 1 is the right default for telemetry and commands; see MQTT QoS levels explained. The Free plan supports QoS up to 1.
5. Expose health. Add Spring Boot Actuator and alert when the MQTT connection is down for longer than expected.

Troubleshooting

The full adapter reference is in the Spring Integration MQTT documentation.

Related Guides

If you need a plain Java client without Spring, the Java MQTT client tutorial compares Paho with the HiveMQ client. The Go MQTT client tutorial builds the same pattern in Go, and the MQTT client libraries guide covers other ecosystems.

Connect Spring Boot to ConnectMQ

A managed MQTT broker gives your Spring Boot service a stable, encrypted endpoint while you focus on the API. ConnectMQ supports MQTT 5 with session expiry and shared subscriptions on paid plans, and its devices view and live topic explorer help you check what your fleet is sending. For a wider look at hosting choices, see the cloud MQTT broker guide. Create a free Serverless broker at app.connectmq.com with no credit card and set the environment variables above.

Frequently Asked Questions

How do I use MQTT in Spring Boot?

Add spring-integration-mqtt and the Paho MQTT 5 client, then configure a Mqttv5PahoMessageDrivenChannelAdapter for inbound messages and a Mqttv5PahoMessageHandler for outbound ones.

Can the inbound and outbound MQTT adapters share one connection?

Yes. Since Spring Integration 6, pass the same Mqttv5ClientManager to both, which uses a single client ID and connection.

Does Spring Integration support MQTT 5?

Yes, through the Mqttv5-prefixed adapters, which are built on the Eclipse Paho MQTT 5 client.

How do I connect Spring Boot to an MQTT broker over TLS?

Use an ssl://host:8883 server URI in the connection options. Brokers with publicly trusted certificates work with the JDK's default trust store.