Spring Boot ve Spring Integration ile MQTT

Spring Boot'un kendine ait bir MQTT istemcisi yoktur, ama Spring Integration'ın vardır: spring-integration-mqtt modülü Eclipse Paho'yu channel adapter'lar içine sarar. Böylece gelen MQTT mesajları bir kanaldaki Spring Message nesnelerine dönüşür, giden bir kanala gönderdiğiniz her şey de broker'a publish edilir. Bu eğitimde subscribe olan, bir gateway arayüzü üzerinden publish eden ve tarayıcıdan uçtan uca test edilebilen küçük bir servis oluşturacaksınız.

Bağımlılıklar

Spring Boot'un bağımlılık yönetimi Spring Integration sürümünü kapsar; sürümü sizin sabitlemeniz gerekmez:

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

Modül, MQTT 3.1.1 için Paho mqttv3 istemcisini getirir. MQTT 5 adapter'larını istiyorsanız ayrıca org.eclipse.paho:org.eclipse.paho.mqttv5.client ekleyin. Aşağıdaki örnekler Spring Boot 3 veya üstünü ve Java 17'yi kullanır.

MQTT birkaç entegrasyon noktasından biriyse ve kanalları, transformer'ları ve hata yönetimini tek bir modelde istiyorsanız Spring Integration doğru seçimdir. Servisiniz yalnızca bir istemciye ihtiyaç duyuyorsa ve reactive streams veya tam MQTT 5 kontrolü istiyorsanız, düz bir Spring bean'ine sarılmış HiveMQ MQTT Client daha hafif bir alternatiftir.

Yapılandırma property'leri

Broker ayarlarını koddan uzak tutun. @ConfigurationProperties ile bağlanan küçük bir record, bunları tipli tutar:

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

Bir yapılandırma sınıfında @EnableConfigurationProperties(MqttProperties.class) ile etkinleştirin. TLS için 8883 portuyla ssl://, yerel geliştirmede şifresiz bağlantılar için 1883 ile tcp:// kullanın.

Client factory

Tüm adapter'lar bağlantılarını bir MqttPahoClientFactory'den alır. TLS, kimlik bilgileri, keep alive ve yeniden bağlanma davranışı burada tek seferde yapılandırılır:

@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 sıradan Paho sınıfıdır; dolayısıyla Last Will, özel SSLSocketFactory, MQTT sürümü gibi tüm Paho seçenekleri kullanılabilir. Her seçeneğin ayrıntıları için Eclipse Paho ile Java'da MQTT yazısına bakın.

Gelen yön: message-driven adapter ile subscribe

MqttPahoMessageDrivenChannelAdapter bir veya birden fazla topic filtresine subscribe olur ve aldığı her mesajı bir çıkış kanalına iletir. Ardından bir @ServiceActivator mesajı işler:

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

Gelen her mesajdaki faydalı header'lar:

Varsayılan converter payload'ları String'e çevirir. CBOR veya Protobuf gibi ikili formatlar alıyorsanız converter.setPayloadAsBytes(true) çağırın. Çalışma zamanında adapter.addTopic(...) ve adapter.removeTopic(...) ile topic ekleyip çıkarabilirsiniz.

Hatalar ve yavaş handler'lar

DirectChannel ile handler'ınız Paho'nun callback thread'inde çalışır. Bunun planlanması gereken iki sonucu vardır:

• Handler içindeki bir exception, adapter.setErrorChannel(...) ile bir hata kanalı ayarladıysanız oraya gönderilir. Ayarlamadıysanız exception loglanır ve adapter sürümüne ve ayarlara bağlı olarak Paho hatayı kopan bağlantı gibi ele alabilir. Hataları her zaman bilinçli olarak yakalayıp yönlendirin; örneğin başarısız payload'ları bir dead letter topic'ine yazan bir kanala.
• Yavaş bir handler, QoS 1 mesajlarının acknowledgement'ını geciktirir ve o bağlantıdaki diğer tüm mesajları bloklar. İşleme bir veritabanı veya HTTP çağrısı içeriyorsa mqttInputChannel'ı sınırlı bir thread pool'a dayanan bir ExecutorChannel ile değiştirin; böylece MQTT thread'i mesajı yalnızca devreder.

Mesajı başka bir thread'e devretmenin, işleminiz bitmeden PUBACK'in gönderilmesi anlamına geldiğini unutmayın. Servis arada çökerse o mesaj kaybolur. At-least-once işleme gerçekten önemliyse adapter.setManualAcks(true) kullanın ve iş kalıcı hale geldikten sonra handler'ınızdan IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK header'ı ile onay verin.

Giden yön: gateway üzerinden publish

MqttPahoMessageHandler giriş kanalına ulaşan her şeyi publish eder. Kodunuzun geri kalanı düz bir Java arayüzü çağırsın diye onu bir @MessagingGateway ile eşleştirin:

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

Artık herhangi bir bean MqttGateway'i inject edip gateway.send("{\"id\":42}", "testmqtt/spring/out/orders") çağırabilir. setAsync(true), gönderimin PUBACK'i beklemeden dönmesini sağlar; setAsyncEvents(true) ile Spring, broker teslimatı onayladığında bir MqttMessageDeliveredEvent yayınlar.

Java DSL alternatifi

Ayrı bean'ler yerine flow'ları tercih ediyorsanız Java DSL, aynı gelen pipeline'ı bir dönüştürme adımı dahil birkaç satırda ifade eder:

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

testmqtt/spring/sensors/+/temp gibi bir filtrenin hangi topic'lerle eşleştiğini deploy etmeden önce topic matcher ile kontrol edin.

Bağlantı olayları ve MQTT 5

Spring Integration, MQTT yaşam döngüsü için application event'leri yayınlar. Bunları dinlemek, bağlantı sorunlarını loglamanın veya alarm üretmenin en basit yoludur:

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

MQTT 5 için sınıfları değiştirin: Mqttv5PahoMessageDrivenChannelAdapter ve Mqttv5PahoMessageHandler, factory yerine Paho v5 MqttConnectionOptions alır ve MQTT 5 user properties'i mesaj header'larına ve header'lardan eşler. Kanallar, gateway ve handler'lar aynı kalır.

Shared subscription ile ölçekleme

Normal bir abonelikle servisinizin her instance'ı her mesajı alır; üç pod, her siparişin üç kez işlenmesi demektir. MQTT 5 shared subscription'ları (birçok broker bunu MQTT 3.1.1 bağlantılarında da uzantı olarak destekler) bu sorunu çözer: $share/orders/testmqtt/spring/in/# topic'ine subscribe olun, broker mesajları orders grubunun tüm üyelerine dağıtsın. application.yml içinde yalnızca topic değişir. Teslim kuralları için MQTT shared subscription'lar yazısını okuyun.

Uçtan uca test edin

1. Uygulamayı başlatın ve logda aboneliği takip edin.
2. Online MQTT istemcisini açın, aynı broker'a bağlanın ve testmqtt/spring/in/demo topic'ine hello publish edin. Servisiniz received testmqtt/spring/in/demo: hello loglamalıdır.
3. Tarayıcıda testmqtt/spring/out/# topic'ine subscribe olun, MqttGateway kullanan bir endpoint'i çağırın ve mesajın QoS 1 ile göründüğünü kontrol edin.
4. Broker bağlantısını kesin (örneğin ağı kısa süre devre dışı bırakarak) ve bir MqttConnectionFailedEvent loglandığını ve adapter'ın yeniden bağlandığını doğrulayın.

Spring Boot servisinizin trafiğini izleyin

Mesajlar gelmiyorsa logdaki Paho exception'ı genellikle nedenini söyler: Paho Java hataları en yaygın olanları listeler, reason code tablosu da broker reddedişlerini çözer.

Spring Integration, MQTT'yi Spring Boot uygulamanızda sıradan bir kanala dönüştürür: client factory'yi bir kez yapılandırın, gelen adapter'ları service activator'lara bağlayın, bir gateway üzerinden publish edin ve client ID'leri bağlantı başına benzersiz tutun. Tarayıcı testi geçtiğinde aynı yapılandırma, yalnızca kimlik bilgileri değiştirilerek production'da çalışır.

Sıkça sorulan sorular