Spring Boot MQTT Entegrasyonu: Cihaz Verisini Tüketin ve REST ile Sunun

Spring Boot MQTT servisi, bir cihaz filosunu iş sistemlerinizin geri kalanına bağlamanın yaygın bir yoludur: cihazlar MQTT üzerinden yayın yapar, web uygulamalarınız ve iş ortaklarınız ise tanıdık bir REST API kullanır. Bu eğitimde Spring Integration'ın MQTT 5 adaptörlerini kullanarak telemetriyi TLS üzerinden tüketecek, her cihazın son okumasını tutacak, bunu /api/devices adresinde sunacak ve cihazlara komut göndereceksiniz; hepsi tek bir paylaşılan MQTT bağlantısı üzerinden.

Mimari

Servis iki dünyanın arasında yer alır:

Üretim ortamı için en önemli ayrıntı bağlantıdır. Inbound adaptör ve outbound handler her biri kendi MQTT client'ını oluşturursa iki client ID'ye ve iki bağlantıya ihtiyaç duyarlar. ConnectMQ dahil pek çok barındırılan broker, bir kimlik bilgisi setine bağlı tek bir client ID verir; bu nedenle aynı ID'ye sahip iki bağlantı birbirini sürekli olarak düşürür. Spring Integration 6 bunu her iki bileşenin paylaştığı bir ClientManager ile çözer. Bu eğitim onu en baştan kullanır.

Proje Kurulumu

pom.xml dosyasında şu bağımlılıklarla bir Spring Boot 3 projesi (Java 17+) oluşturun:

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

Paho MQTT 5 client, spring-integration-mqtt için isteğe bağlı bir bağımlılıktır; bu yüzden onu açıkça eklemeniz gerekir. Bunu unutmak, başlangıçta ClassNotFoundException: org.eclipse.paho.mqttv5... hatasının en yaygın nedenidir.

Yapılandırma

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}

Varsayılan değerler, boş olmayan ve benzersiz herhangi bir client ID ile anonim istemcileri kabul eden broker.connectmq.com adresindeki ücretsiz herkese açık MQTT broker'ı gösterir. Herkese açık alandaki topic'ler herkese görünür; bu yüzden rastgele bir önek ve yalnızca test verisi kullanın. 8883 portuyla birlikte ssl:// şeması TLS'yi etkinleştirir; broker'ın Let's Encrypt sertifikasına JDK kutudan çıktığı haliyle güvenir.

MQTT Yapılandırma Sınıfı

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

Hem adaptör hem de handler aynı Mqttv5ClientManager örneğini alır; böylece uygulama tek bir client ID ile tam olarak bir MQTT bağlantısı açar. Otomatik yeniden bağlanma bağlantı seçeneklerinde yer alır; setCleanStart(false) ve bir oturum sona erme aralığıyla broker, kısa ağ kesintilerinde aboneliği korur.

Outbound Gateway

Bir messaging gateway, uygulamanın geri kalanına yayın yapmak için sade bir Java arayüzü sunar:

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

v5 handler, String payload'ı UTF-8 byte'lara dönüştürür; MqttHeaders.TOPIC header'ı ise her mesaj için hedef topic'i belirler.

Son Okumaları Saklamak

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

Inbound adaptör varsayılan olarak byte[] payload'lar iletir ve alınan topic MqttHeaders.RECEIVED_TOPIC header'ında gelir. Cihaz ID'si, telemetry öncesindeki segmenttir.

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

ID kontrolü süs değildir. Bu kontrol olmadan bir çağıran, /, + veya # içeren bir ID gönderip hiç açmayı düşünmediğiniz topic'lere yayın yapabilir.

Bir Mesaj Uygulamada Nasıl Akar?

Tek bir telemetri mesajını baştan sona izlemek faydalıdır, çünkü Spring Integration birkaç bean'in arkasında pek çok altyapıyı gizler.

1. Cihaz {"temperature":22.5} mesajını <prefix>/devices/sensor-1/telemetry topic'ine yayınlar.
2. Broker mesajı, paylaşılan client manager aracılığıyla inbound adaptörün oluşturduğu abonelik dahil, eşleşen tüm aboneliklere yönlendirir.
3. Adaptör payload byte'larını bir Spring Message içine sarar ve RECEIVED_TOPIC, RECEIVED_QOS ve RECEIVED_RETAINED gibi header'lar ekler.
4. telemetryFlow, mesajı JSON'u ayrıştırıp o cihazın önceki okumasını değiştiren ReadingStore.accept metoduna iletir.
5. /api/devices/sensor-1 adresine yapılan bir REST çağrısı map'i okur ve son okumayı JSON olarak döndürür.

Komutlar ters yönde ilerler: controller gateway'i çağırır, gateway mqttOutbound kanalına bir mesaj gönderir, service activator bunu outbound handler'a verir ve handler aynı paylaşılan bağlantı üzerinden yayın yapar. Outbound handler asenkron olduğu için REST çağrısı broker onayını beklemeden 202 Accepted döndürür. API'nizin teslimatı doğrulaması gerekiyorsa handler'ı senkron moda alın ve ek gecikmeyi kabul edin.

Hata yönetimi

Varsayılan olarak akış içindeki bir istisna loglanır ve mesaj atılır. Bir prototipin ötesindeki her şey için inbound adaptöre adapter.setErrorChannelName("mqttErrors") ile bir hata kanalı ve topic ile payload'ı loglayan küçük bir akış ekleyin; böylece bir cihazdan gelen hatalı biçimli bir mesaj sessizce yok sayılmak yerine görünür olur. Handler'ı hızlı tutun: veritabanı yazma işlemleri eklerseniz, yavaş bir veritabanının mesaj teslimatını bekletmemesi için adaptör ile store arasında bir executor channel kullanmayı düşünün.

Uçtan Uca Test

Uygulamayı ./mvnw spring-boot:run ile başlatın, ardından bir cihazı simüle edin:

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

REST API'yi sorgulayın:

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

Bir komut gönderin ve cihaz topic'ine ulaşmasını izleyin:

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

Spring Boot MQTT için Üretim Notları

1. Verilen kimlik bilgilerini kullanın. ConnectMQ portalında bir broker oluşturun ve MQTT_CLIENT_ID, MQTT_USERNAME ile MQTT_PASSWORD değişkenlerini üretilen değerlere ayarlayın; client ID birebir eşleşmelidir.
2. Bir örnek, bir client ID. Birkaç replikaya ölçeklenirseniz her birinin kendi kimlik bilgilerine ihtiyacı olur. Aynı telemetrinin yük dengeli tüketimi için MQTT 5 shared subscription'lar ücretli Serverless planlarda mevcuttur.
3. Önemli olanı kalıcı hale getirin. Bellek içi map yeniden başlatmada kaybolur. Geçmiş için okumaları akış içinde bir veritabanına yazın veya bu servisin yanına özel bir veri alım aracı abone edin.
4. QoS'u bilinçli seçin. Telemetri ve komutlar için QoS 1 doğru varsayılandır; MQTT QoS seviyeleri açıklaması yazısına bakın. Free plan QoS 1'e kadar destekler.
5. Sağlık durumunu sunun. Spring Boot Actuator ekleyin ve MQTT bağlantısı beklenenden uzun süre kapalı kaldığında uyarı verin.

Sorun Giderme

Adaptörlerin tam başvurusu Spring Integration MQTT belgelerinde yer alır.

İlgili Rehberler

Spring olmadan sade bir Java client'a ihtiyacınız varsa Java MQTT client eğitimi Paho ile HiveMQ client'ı karşılaştırır. Go MQTT istemcisi eğitimi aynı deseni Go ile kurar; MQTT istemci kütüphaneleri rehberi ise diğer ekosistemleri ele alır.

Spring Boot'u ConnectMQ'ya Bağlayın

Yönetilen bir MQTT broker, siz API'ye odaklanırken Spring Boot servisinize kararlı ve şifreli bir uç nokta sağlar. ConnectMQ, ücretli planlarda oturum sona erme süresi ve shared subscription'larla MQTT 5'i destekler; cihazlar görünümü ve canlı topic gezgini filonuzun ne gönderdiğini kontrol etmenize yardımcı olur. Barındırma seçeneklerine daha geniş bir bakış için bulut MQTT broker rehberine göz atın. app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun ve yukarıdaki ortam değişkenlerini ayarlayın.

Sık Sorulan Sorular

Spring Boot'ta MQTT nasıl kullanılır?

spring-integration-mqtt ve Paho MQTT 5 client'ı ekleyin; ardından gelen mesajlar için bir Mqttv5PahoMessageDrivenChannelAdapter, giden mesajlar için bir Mqttv5PahoMessageHandler yapılandırın.

Inbound ve outbound MQTT adaptörleri tek bir bağlantıyı paylaşabilir mi?

Evet. Spring Integration 6'dan itibaren her ikisine de aynı Mqttv5ClientManager örneğini verin; bu, tek bir client ID ve bağlantı kullanır.

Spring Integration MQTT 5'i destekliyor mu?

Evet; Eclipse Paho MQTT 5 client üzerine kurulu, Mqttv5 önekli adaptörler aracılığıyla.

Spring Boot'u bir MQTT broker'a TLS üzerinden nasıl bağlarım?

Bağlantı seçeneklerinde bir ssl://host:8883 sunucu URI'si kullanın. Herkesçe güvenilen sertifikalara sahip broker'lar JDK'nın varsayılan trust store'u ile çalışır.