HiveMQ MQTT Client ile Java'da MQTT

HiveMQ MQTT Client, MQTT 3.1.1 ve MQTT 5 için üç API stili sunan açık kaynak bir Java kütüphanesidir: blocking, asenkron (CompletableFuture) ve reactive (RxJava; isteğe bağlı bir Reactor modülüyle). Adına rağmen standartlara uyan her broker ile çalışır. Bu eğitimde istemciyi TLS üzerinden bağlayacak, her API stiliyle publish ve subscribe yapacak, MQTT 5 özelliklerini kullanacak ve her şeyi dışarıdan doğrulayacaksınız.

Bağımlılığı ekleyin

İstemci Java 8 veya üstünü gerektirir. Gradle ile:

dependencies {
    implementation("com.hivemq:hivemq-mqtt-client:1.3.3")
    // Only if you connect over WebSockets:
    // implementation("com.hivemq:hivemq-mqtt-client-websocket:1.3.3")
}

Maven ile:

<dependency>
  <groupId>com.hivemq</groupId>
  <artifactId>hivemq-mqtt-client</artifactId>
  <version>1.3.3</version>
</dependency>

En yeni 1.3.x sürümü için Maven Central'a bakın. WebSocket transport ve Reactor API gibi isteğe bağlı özellikler ayrı artifact'lerde bulunur; böylece çekirdek jar küçük kalır.

İstemci oluşturun

Her istemci MqttClient.builder() ile başlar. Builder değiştirilemez ve fluent'tir: protokol sürümünü, client identifier'ı, sunucuyu ve transport'u seçer, ardından buildBlocking(), buildAsync() veya buildRx() çağırırsınız.

import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
import java.util.UUID;

Mqtt5BlockingClient client = MqttClient.builder()
        .useMqttVersion5()
        .identifier("testmqtt-java-" + UUID.randomUUID().toString().substring(0, 8))
        .serverHost("broker.connectmq.com")
        .serverPort(8883)
        .sslWithDefaultConfig()                 // JVM trust store ile TLS
        .automaticReconnectWithDefaultConfig()  // üstel backoff
        .buildBlocking();

Baştan bilmeniz gereken birkaç nokta:

• Client identifier'lar her bağlantı için benzersiz olmalıdır. Aynı ID'ye sahip iki süreç birbirini döngü halinde bağlantıdan düşürür. Client ID oluşturucu ile bir tane üretin ya da kararlı bir cihaz veya instance adından türetin.
• sslWithDefaultConfig(), genel CA'lara zaten güvenen JVM varsayılan trust store'unu kullanır. Özel bir CA için .sslConfig().trustManagerFactory(tmf).applySslConfig() ile bir TrustManagerFactory verin.
• MQTT 3.1.1 için useMqttVersion3() kullanın; kodun geri kalanı MQTT 5 seçenekleri dışında neredeyse aynıdır.

Kimlik bilgileriyle bağlanın

Bağlantı siz bir CONNECT gönderene kadar açılmaz. Blocking API'de send() CONNACK döndürür ya da exception fırlatır:

import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException;
import java.nio.charset.StandardCharsets;

try {
    client.connectWith()
            .simpleAuth()
                .username(System.getenv("MQTT_USER"))
                .password(System.getenv("MQTT_PASS").getBytes(StandardCharsets.UTF_8))
                .applySimpleAuth()
            .cleanStart(false)
            .sessionExpiryInterval(3600)   // oturumu çevrimdışıyken bir saat tut
            .keepAlive(30)
            .send();
    System.out.println("connected");
} catch (Mqtt5ConnAckException e) {
    System.err.println("rejected: " + e.getMqttMessage().getReasonCode());
}

Broker bağlantıyı reddederse exception, MQTT 5 reason code'unu taşır; örneğin BAD_USER_NAME_OR_PASSWORD veya NOT_AUTHORIZED. Değeri MQTT reason code tablosunda arayın ya da her birinin ne anlama geldiğini görmek için CONNACK reason code'ları açıklaması yazısını okuyun.

Blocking API: publish ve mesaj alma

Blocking API okunması en kolay olanıdır; komut satırı araçları ve testler için idealdir. Mesaj almak için subscribe etmeden önce bir Publishes stream'i açarsınız, böylece hiçbir mesaj kaçmaz:

import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

try (Mqtt5BlockingClient.Mqtt5Publishes publishes =
             client.publishes(MqttGlobalPublishFilter.ALL)) {

    client.subscribeWith()
            .topicFilter("testmqtt/java/demo/#")
            .qos(MqttQos.AT_LEAST_ONCE)
            .send();

    client.publishWith()
            .topic("testmqtt/java/demo/hello")
            .qos(MqttQos.AT_LEAST_ONCE)
            .payload("hello from Java".getBytes(StandardCharsets.UTF_8))
            .send();

    Optional<Mqtt5Publish> msg = publishes.receive(10, TimeUnit.SECONDS);
    msg.ifPresent(p -> System.out.println(
            p.getTopic() + " -> " + new String(p.getPayloadAsBytes(), StandardCharsets.UTF_8)));
}
client.disconnect();

Stream bir kaynak olduğu için try-with-resources bloğu onu temiz biçimde kapatır. Bir testte, timeout sonrası boş bir Optional net bir başarısızlık sinyalidir.

Async API: CompletableFuture zincirleri

Servislerde async API thread'leri bloklamayı önler. Her işlem bir CompletableFuture döndürür; böylece bağlanma, subscribe ve publish adımlarını zincirleyebilirsiniz:

import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;

Mqtt5AsyncClient async = client.toAsync();   // aynı bağlantı, farklı API

async.subscribeWith()
        .topicFilter("testmqtt/java/sensors/+/temp")
        .qos(MqttQos.AT_LEAST_ONCE)
        .callback(p -> System.out.println(
                p.getTopic() + " " + new String(p.getPayloadAsBytes(), StandardCharsets.UTF_8)))
        .send()
        .thenCompose(subAck -> async.publishWith()
                .topic("testmqtt/java/sensors/lab/temp")
                .qos(MqttQos.AT_LEAST_ONCE)
                .payload("21.7".getBytes(StandardCharsets.UTF_8))
                .send())
        .whenComplete((result, err) -> {
            if (err != null) err.printStackTrace();
            else System.out.println("published");
        });

subscribeWith()'e verilen callback, istemcinin Netty event loop'unda çalışır. Kısa tutun: veritabanı yazma, HTTP çağrısı gibi ağır işleri kendi executor'unuza devredin; aksi halde o bağlantıdaki diğer tüm mesajları yavaşlatırsınız. Topic filtrelerini güvenle seçmek için önce topic matcher ile deneyin.

Reactive API: backpressure'lı stream'ler

Mqtt5RxClient abonelikleri RxJava Flowable olarak sunar. Tüketici üreticiden yavaş olabildiğinde bu işe yarar; çünkü backpressure MQTT akış kontrolüne kadar iletilir:

import com.hivemq.client.mqtt.mqtt5.Mqtt5RxClient;

Mqtt5RxClient rx = client.toRx();

rx.subscribePublishesWith()
        .topicFilter("testmqtt/java/events/#")
        .qos(MqttQos.AT_LEAST_ONCE)
        .applySubscribe()
        .doOnSingle(subAck -> System.out.println("subscribed: " + subAck.getReasonCodes()))
        .map(p -> new String(p.getPayloadAsBytes(), StandardCharsets.UTF_8))
        .subscribe(System.out::println, Throwable::printStackTrace);

Stack'iniz Project Reactor kullanıyorsa (örneğin Spring WebFlux), hivemq-mqtt-client-reactor artifact'ini ekleyin ve Flux ile Mono kullanın.

MQTT 5 özellikleri

Builder'lar MQTT 5 property'lerini doğrudan sunar; ham property map'leriyle uğraşmanız gerekmez:

import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PayloadFormatIndicator;

async.publishWith()
        .topic("testmqtt/java/orders/created")
        .qos(MqttQos.AT_LEAST_ONCE)
        .payload("{\"id\":42}".getBytes(StandardCharsets.UTF_8))
        .contentType("application/json")
        .payloadFormatIndicator(Mqtt5PayloadFormatIndicator.UTF_8)
        .messageExpiryInterval(60)                    // 60 sn içinde teslim edilmezse at
        .responseTopic("testmqtt/java/orders/replies")
        .correlationData("req-42".getBytes(StandardCharsets.UTF_8))
        .userProperties()
            .add("source", "billing-service")
            .add("schema", "v2")
            .applyUserProperties()
        .send();

Daha fazlası için MQTT user properties ve MQTT 5 özelliklerini test etmek yazılarına bakın.

Yeniden bağlanma ve yaşam döngüsü listener'ları

automaticReconnectWithDefaultConfig() üstel backoff ile yeniden dener. Bunu ayarlamak ve durum değişikliklerini loglamak için açıkça yapılandırın:

Mqtt5AsyncClient resilient = MqttClient.builder()
        .useMqttVersion5()
        .identifier("testmqtt-java-worker-1")
        .serverHost("broker.connectmq.com")
        .serverPort(8883)
        .sslWithDefaultConfig()
        .automaticReconnect()
            .initialDelay(1, TimeUnit.SECONDS)
            .maxDelay(30, TimeUnit.SECONDS)
            .applyAutomaticReconnect()
        .addConnectedListener(ctx -> System.out.println("connected"))
        .addDisconnectedListener(ctx -> System.out.println("disconnected: " + ctx.getCause().getMessage()))
        .buildAsync();

Yeniden bağlanmayı cleanStart(false) ve bir oturum süresi dolum aralığıyla birleştirin: broker böylece aboneliklerinizi saklar ve istemci yokken QoS 1/2 mesajlarını kuyruğa alır. Bunun için kararlı bir client identifier şarttır; her başlangıçta rastgele bir ID, her seferinde yeni ve boş bir oturum demektir. Paylaşımlı test endpoint'i broker.connectmq.com çevrimdışı oturumları saklamaz; oturum kalıcılığı testlerini kendi özel broker instance'ınızda çalıştırın.

WebSocket ve temiz kapanış

Bazı ağlar yalnızca HTTPS trafiğine izin verir. Classpath'te hivemq-mqtt-client-websocket artifact'i varsa aynı istemci MQTT'yi güvenli WebSocket üzerinden tünelleyebilir; yalnızca transport ayarları değişir:

Mqtt5BlockingClient wsClient = MqttClient.builder()
        .useMqttVersion5()
        .identifier("testmqtt-java-ws")
        .serverHost("broker.connectmq.com")
        .serverPort(8084)
        .sslWithDefaultConfig()
        .webSocketConfig()
            .serverPath("mqtt")
            .applyWebSocketConfig()
        .buildBlocking();

Uygulamanız durduğunda JVM'den öylece çıkmak yerine disconnect() çağırın. Temiz bir DISCONNECT, broker'a Will mesajınızı yayınlamamasını söyler; MQTT 5 ile oturuma artık ihtiyaç yoksa o anda disconnectWith().sessionExpiryInterval(0).send() ile oturum süresini bile değiştirebilirsiniz. Spring veya Jakarta EE uygulamasında bunu container'ın kapanış callback'ine bağlayın; böylece kademeli deploy'lar hayalet oturumlar bırakmaz ve sahte "offline" alarmları tetiklemez.

Dışarıdan test edin

Çalıştığını anlamak için yalnızca kendi istemcinize güvenmeyin. Online MQTT istemcisini açın, aynı broker'a bağlanın ve testmqtt/java/# topic'ine subscribe olun. Her örneği çalıştırın ve mesajların beklenen topic, QoS ve MQTT 5 property'leriyle geldiğini doğrulayın. Ardından tarayıcıdan testmqtt/java/sensors/lab/temp topic'ine publish edin ve async callback'inizin bunu yazdırmasını izleyin.

Java istemcinizin mesajlarını canlı izleyin

Otomatik kontroller için blocking API, JUnit ile çok iyi eşleşir: JUnit 5 ile MQTT entegrasyon testleri yazısına bakın. Hâlâ kütüphane seçiyorsanız Java MQTT istemcileri karşılaştırması bu istemciyi Eclipse Paho ile yan yana koyar.

Sık yapılan hatalar

• Blocking örnekte hiçbir şey gelmiyor — publishes() stream'i abonelikten sonra açılmış ya da global filtre eşleşmiyor. Önce onu açın.
• SSLHandshakeException — genellikle trust store'da eksik bir özel CA ya da şifresiz 1883 portuna TLS ile bağlanma. TLS için 8883 kullanın.
• Callback'ler takılıyor — callback içindeki yavaş kod event loop'u bloklar. Başka bir executor'a taşıyın.
• Yeniden başlatmada oturum kayboluyor — rastgele client ID veya cleanStart(true). Kararlı bir ID ve oturum süresi dolum aralığı kullanın.

HiveMQ MQTT Client, Java'ya tercih ettiğiniz API stilinde, MQTT 5'i tam destekleyen modern bir istemci sunar. Bağlantıyı kanıtlamak için blocking API ile başlayın, production kodu için async veya reactive'e geçin ve geliştirme sırasında harici bir istemciyi açık tutun; böylece broker'dan gerçekte neyin geçtiğini her zaman görürsünüz.

Sıkça sorulan sorular