Java MQTT Client Eğitimi: TLS ve MQTT 5 ile Eclipse Paho ve HiveMQ Client Karşılaştırması

Bir Java MQTT client seçmek genellikle iki kütüphaneye indirgenir: Eclipse Paho ve HiveMQ MQTT Client. Bu eğitim ikisini dürüstçe karşılaştırır, ardından aynı publish/subscribe programını her biriyle 8883 portunda TLS üzerinden MQTT 5, otomatik yeniden bağlanma ve Last Will kullanarak uygular; böylece kod tabanınıza uygun olanı seçebilirsiniz.

Java MQTT Client Kütüphanelerinin Karşılaştırması

Her iki kütüphane de açık kaynaklıdır, ikisi de MQTT 5 konuşur ve ikisi de üretim ortamında kullanılır. Esas olarak API tarzında ve ne kadar aktif geliştirildiklerinde ayrışırlar.

Pratik bir kural: Paho küçüktür, bağımlılık içermez ve sayısız örnekten tanıdıktır; bu da onu Android veya kısıtlı ortamlar için cazip kılar. HiveMQ client daha modern bir API'ye sahiptir ve yüksek mesaj hızlarında geri basıncı (back-pressure) iyi yönetir; bu da backend servislerine uyar. Spring geliştiricileri Paho'yu çoğunlukla Spring Integration aracılığıyla dolaylı olarak kullanır; bu yol için Spring Boot MQTT rehberine bakın.

Proje Kurulumu

Her iki örnek de Java 17 ve Maven kullanır. pom.xml dosyasına bağımlılıklardan birini veya ikisini ekleyin:

<dependencies>
  <dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.mqttv5.client</artifactId>
    <version>1.2.5</version>
  </dependency>
  <dependency>
    <groupId>com.hivemq</groupId>
    <artifactId>hivemq-mqtt-client</artifactId>
    <version>1.3.3</version>
  </dependency>
</dependencies>

Test için broker.connectmq.com adresindeki herkese açık MQTT test broker ile bağlanın. Boş olmayan ve benzersiz herhangi bir client ID ile anonim istemcileri kabul eder. connectmq/demo/j6t1ve gibi rastgele bir önek kullanın ve herkese açık topic'lerin herkes tarafından okunabildiğini unutmayın.

Java'nın varsayılan trust store'u, tüm güncel JDK sürümlerinde broker'ın kullandığı Let's Encrypt kökü ISRG Root X1'i içerir. TLS için keystore kurulumu gerekmez.

Örnek 1: Eclipse Paho MQTT 5 Client

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

public class PahoDemo {

    static String env(String key, String def) {
        String v = System.getenv(key);
        return (v == null || v.isEmpty()) ? def : v;
    }

    public static void main(String[] args) throws Exception {
        String serverUri = env("MQTT_URL", "ssl://broker.connectmq.com:8883");
        String clientId = env("MQTT_CLIENT_ID", "paho-j6t1ve");
        String prefix = env("MQTT_PREFIX", "connectmq/demo/j6t1ve");
        String user = System.getenv("MQTT_USERNAME");
        String commandFilter = prefix + "/cmd/#";
        ExecutorService worker = Executors.newSingleThreadExecutor();

        MqttClient client = new MqttClient(serverUri, clientId, new MemoryPersistence());

        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setAutomaticReconnect(true);
        options.setCleanStart(false);
        options.setSessionExpiryInterval(600L);
        options.setKeepAliveInterval(30);
        if (user != null) {
            options.setUserName(user);
            options.setPassword(env("MQTT_PASSWORD", "").getBytes(StandardCharsets.UTF_8));
        }
        MqttMessage will = new MqttMessage("offline".getBytes(StandardCharsets.UTF_8));
        will.setQos(1);
        will.setRetained(true);
        options.setWill(prefix + "/status", will);

        client.setCallback(new MqttCallback() {
            @Override
            public void connectComplete(boolean reconnect, String uri) {
                System.out.println((reconnect ? "reconnected to " : "connected to ") + uri);
                if (reconnect) {
                    // Callback thread'ini asla bloklamayın; yeniden aboneliği bir worker üzerinde yapın.
                    worker.submit(() -> {
                        try {
                            client.subscribe(commandFilter, 1);
                        } catch (MqttException e) {
                            System.err.println("resubscribe failed: " + e.getMessage());
                        }
                    });
                }
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                System.out.println("command on " + topic + ": "
                        + new String(message.getPayload(), StandardCharsets.UTF_8));
            }

            @Override
            public void disconnected(MqttDisconnectResponse response) {
                System.out.println("disconnected: " + response.getReasonString());
            }

            @Override
            public void mqttErrorOccurred(MqttException exception) {
                System.err.println("error: " + exception.getMessage());
            }

            @Override
            public void deliveryComplete(IMqttToken token) { }

            @Override
            public void authPacketArrived(int reasonCode, MqttProperties properties) { }
        });

        client.connect(options);
        client.subscribe(commandFilter, 1);

        MqttMessage online = new MqttMessage("online".getBytes(StandardCharsets.UTF_8));
        online.setQos(1);
        online.setRetained(true);
        client.publish(prefix + "/status", online);

        while (true) {
            String json = String.format("{\"device\":\"%s\",\"temperature\":%.2f,\"ts\":\"%s\"}",
                    clientId, 20 + Math.random() * 5, Instant.now());
            MqttMessage msg = new MqttMessage(json.getBytes(StandardCharsets.UTF_8));
            msg.setQos(1);
            try {
                client.publish(prefix + "/telemetry", msg);
            } catch (MqttException e) {
                System.err.println("publish failed: " + e.getMessage());
            }
            Thread.sleep(5000);
        }
    }
}

Önemli noktalar: ssl:// şeması TLS'yi etkinleştirir; setAutomaticReconnect(true) bağlantı kopmalarını yönetir; oturum sona erme süresiyle birlikte setCleanStart(false) kısa kesintilerde abonelikleri korur. connectComplete içinde yeniden abone olmak daha uzun kesintileri kapsar; ancak bir Paho callback'inden bloklayan client metotlarını çağırmak client'ı durdurabileceği için bu işlem bir worker thread üzerinde çalışır.

Örnek 2: HiveMQ MQTT Client

import java.nio.charset.StandardCharsets;
import java.time.Instant;

import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;

public class HiveMqDemo {

    static String env(String key, String def) {
        String v = System.getenv(key);
        return (v == null || v.isEmpty()) ? def : v;
    }

    public static void main(String[] args) throws Exception {
        String host = env("MQTT_HOST", "broker.connectmq.com");
        String clientId = env("MQTT_CLIENT_ID", "hivemq-j6t1ve");
        String prefix = env("MQTT_PREFIX", "connectmq/demo/j6t1ve");
        String user = System.getenv("MQTT_USERNAME");

        Mqtt5AsyncClient client = MqttClient.builder()
                .useMqttVersion5()
                .identifier(clientId)
                .serverHost(host)
                .serverPort(8883)
                .sslWithDefaultConfig()
                .automaticReconnectWithDefaultConfig()
                .addConnectedListener(ctx -> System.out.println("connected"))
                .addDisconnectedListener(ctx ->
                        System.out.println("disconnected: " + ctx.getCause().getMessage()))
                .buildAsync();

        var connect = client.connectWith()
                .cleanStart(false)
                .sessionExpiryInterval(600)
                .keepAlive(30)
                .willPublish()
                    .topic(prefix + "/status")
                    .payload("offline".getBytes(StandardCharsets.UTF_8))
                    .qos(MqttQos.AT_LEAST_ONCE)
                    .retain(true)
                    .applyWillPublish();
        if (user != null) {
            connect = connect.simpleAuth()
                    .username(user)
                    .password(env("MQTT_PASSWORD", "").getBytes(StandardCharsets.UTF_8))
                    .applySimpleAuth();
        }
        connect.send().join();

        client.subscribeWith()
                .topicFilter(prefix + "/cmd/#")
                .qos(MqttQos.AT_LEAST_ONCE)
                .callback(publish -> System.out.println("command on " + publish.getTopic() + ": "
                        + new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8)))
                .send()
                .join();

        client.publishWith()
                .topic(prefix + "/status")
                .payload("online".getBytes(StandardCharsets.UTF_8))
                .qos(MqttQos.AT_LEAST_ONCE)
                .retain(true)
                .send();

        while (true) {
            String json = String.format("{\"device\":\"%s\",\"temperature\":%.2f,\"ts\":\"%s\"}",
                    clientId, 20 + Math.random() * 5, Instant.now());
            client.publishWith()
                    .topic(prefix + "/telemetry")
                    .payload(json.getBytes(StandardCharsets.UTF_8))
                    .qos(MqttQos.AT_LEAST_ONCE)
                    .contentType("application/json")
                    .messageExpiryInterval(60)
                    .send()
                    .whenComplete((result, error) -> {
                        if (error != null) System.err.println("publish failed: " + error.getMessage());
                    });
            Thread.sleep(5000);
        }
    }
}

Builder neredeyse bir cümle gibi okunur. sslWithDefaultConfig() JDK trust store'u ile TLS'yi etkinleştirir, automaticReconnectWithDefaultConfig() ise üstel geri çekilme (exponential back-off) ile yeniden bağlanır. MQTT 5 özellikleri contentType ve messageExpiryInterval, tüketicilere payload formatını bildirir ve broker'ın bayat okumaları bir dakika sonra atmasını sağlar. MQTT 5 özelliklerine genel bakış bu özellikleri daha ayrıntılı açıklar.

Bloklayan, Asenkron mu, Reaktif mi?

Her iki örnek de akışın kolay izlenmesi için Thread.sleep içeren basit bir while döngüsü kullanır; ancak gerçek servisler nadiren böyle görünür. Seçtiğiniz API tarzı kodun geri kalanını şekillendirir, bu yüzden erken karar vermeye değer.

Bloklayan çağrılar okunması en kolay olanlardır: connect, subscribe ve publish yalnızca broker yanıt verdiğinde döner. Paho'nun MqttClient sınıfı ve HiveMQ Mqtt5BlockingClient bu şekilde çalışır. Bağlantı başına bir thread'in kabul edilebilir olduğu komut satırı araçlarına, testlere ve küçük ağ geçitlerine uyarlar.

Asenkron çağrılar bir token (Paho'nun MqttAsyncClient sınıfında IMqttToken) veya bir CompletableFuture (HiveMQ) ile hemen döner. Yüksek hızda yayın yapan backend servislerine uyarlar, çünkü yavaş bir onay asla üretici thread'ini durdurmaz. HiveMQ örneğinin whenComplete ile yaptığı gibi, hata yönetimini görmezden gelmek yerine future'a zincirleyin.

Reaktif akışlar, yalnızca HiveMQ client'ta RxJava API'si üzerinden mevcuttur ve gelen mesajları gerçek geri basınçlı bir Flowable olarak modeller. Uygulamanız zaten Reactor veya RxJava kullanıyorsa bu, tüketiciler geride kaldığında mesajların bellekte tamponlanmasını önler.

Hangisini seçerseniz seçin, mesaj işlemeyi ağ thread'inin dışında tutun. messageArrived veya bir HiveMQ callback'i içindeki bir veritabanı ekleme ya da HTTP çağrısı, o bağlantıdaki diğer tüm mesajları geciktirir. Payload'ı bir kuyruğa veya executor'a aktarın ve hızlıca dönün.

Java Servisleri için Topic Tasarımı

Örnekler üç topic ailesi kullanır: okumalar için .../telemetry, komutlar için .../cmd/# ve erişilebilirlik için .../status. Komutları ve veriyi ayrı tutmak, bir tüketicinin komutları hiç görmeden tek bir joker karakterle telemetriye abone olabilmesi ve izinlerin daha sonra topic ailesine göre değerlendirilebilmesi anlamına gelir. Saha, hat ve cihaz gibi filtreleme yapacağınız tanımlayıcıları yalnızca JSON payload'a değil topic seviyelerine de koyun, çünkü broker yalnızca topic'lere göre yönlendirme yapabilir.

Her İki Client'ı Test Etmek

Sınıflardan birini çalıştırın ve topic'leri bir terminalden izleyin:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-j6t1ve -t 'connectmq/demo/j6t1ve/#' -v
mosquitto_pub -h broker.connectmq.com -p 1883 -i send-j6t1ve -t 'connectmq/demo/j6t1ve/cmd/valve' -m 'close'

Varsayılan ID'lerle aynı anda yalnızca bir client çalıştırın veya her birine kendi ID'sini verin; iki client bir oturumu paylaşamaz.

Özel Bir Broker'a Geçiş

Üretim ortamı için ConnectMQ portalında bir broker oluşturun ve üretilen kullanıcı adını, parolayı ve client ID'yi kullanın. Client ID verilen kimlik bilgilerine bağlıdır; bu nedenle birebir eşleşmelidir:

export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_PREFIX=warehouse/zone3/gateway

Paho tam URI'yi MQTT_URL değişkeninden okur (ssl://broker.connectmq.com:8883); HiveMQ örneği ise MQTT_HOST değişkenini okur.

Java MQTT Client Kontrol Listesi

1. Her JVM için benzersiz client ID. Aynı ID'ye sahip birden fazla örnek birbirini çevrimdışı bırakır.
2. Varsayılan olarak QoS 1. Free plan QoS 0 ve 1'i destekler; QoS 2 ücretli bir plan gerektirir. MQTT QoS seviyeleri açıklaması yazısına bakın.
3. Callback'leri bloklamayın. Her iki kütüphanede de işi bir executor'a devredin.
4. Yayın hatalarını ele alın. Bağlantı yokken mesajları atmaya, tamponlamaya veya yeniden denemeye karar verin.
5. Reason code'ları loglayın. MQTT 5 bağlantı kopmaları bir neden taşır; örneğin bir kota aşıldığında 0x97.
6. Kimlik bilgilerini kodun dışında, ortam değişkenlerinde veya bir gizli bilgi deposunda tutun.

Sorun Giderme

Resmi Paho proje sayfası eclipse.dev/paho adresindedir.

Daha Fazla Backend Eğitimi

Ekibiniz Go veya Rust da yazıyorsa Go MQTT istemcisi eğitimi ile karşılaştırın. Diller arası daha geniş bir inceleme için MQTT istemci kütüphaneleri rehberini okuyun.

Java Servislerinizi ConnectMQ'ya Bağlayın

Barındırılan bir MQTT broker, Java servislerinizin işletilecek bir broker olmadan tek bir TLS uç noktasıyla bağlanmasını sağlar. ConnectMQ MQTT 3.1.1 ve 5.0'ı destekler; böylece hem Paho hem de HiveMQ client değişiklik yapmadan çalışır ve portalın canlı topic gezgini kodunuzun tam olarak ne yayınladığını gösterir. Barındırma seçeneklerini hâlâ karşılaştırıyorsanız bulut MQTT broker rehberi artıları ve eksileri ele alır. app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun.

Sık Sorulan Sorular

En iyi Java MQTT client kütüphanesi hangisi?

İkisi de sağlamdır. Paho bağımlılık içermez ve geniş çapta belgelenmiştir; HiveMQ MQTT Client ise modern, akıcı bir API'ye, asenkron ve reaktif desteğe ve daha aktif bir geliştirme sürecine sahiptir.

Eclipse Paho, Java'da MQTT 5'i destekliyor mu?

Evet, ayrı org.eclipse.paho.mqttv5.client artifact'i aracılığıyla. Eski org.eclipse.paho.client.mqttv3 artifact'i yalnızca MQTT 3.1.1'i destekler.

Bir Java MQTT client'ta TLS'yi nasıl etkinleştiririm?

Paho ile bir ssl://host:8883 URI'si kullanın. HiveMQ client ile sslWithDefaultConfig() çağırın ve 8883 portunu kullanın. Her ikisi de herkesçe güvenilen sertifikalar için JDK trust store'una dayanır.

Java MQTT client'lar otomatik olarak yeniden bağlanır mı?

Evet, etkinleştirilirse: Paho'da setAutomaticReconnect(true), HiveMQ client'ta automaticReconnectWithDefaultConfig(). Oturumun sona ermiş olabileceği durumlarda yeniden bağlandıktan sonra tekrar abone olun.