rumqttc ile Rust MQTT: Tokio ile TLS Üzerinden Asenkron Publish/Subscribe

Bu Rust MQTT rumqttc eğitimi, rumqttc'nin uç ağ geçitlerinden backend veri alım servislerine kadar Tokio servislerine doğal olarak uyan, tamamen Rust ile yazılmış asenkron bir istemciyi nasıl sunduğunu gösterir. Bu eğitimde 8883 portunda rustls ile TLS üzerinden bağlanan, JSON telemetri yayınlayan, komut alan ve ağ kesintilerinde çalışmaya devam eden eksiksiz bir rumqttc istemcisi geliştireceksiniz.

Rust MQTT için Neden rumqttc?

rumqttc, rumqtt projesinin istemci tarafıdır. Tamamen Rust ile yazılmıştır; crate kökünde MQTT 3.1.1'i, v5 modülünde ise MQTT 5'i destekler ve hem senkron bir Client hem de Tokio üzerine kurulu asenkron bir AsyncClient sunar. TLS varsayılan olarak rustls'den gelir; bu nedenle hedef makinede OpenSSL gerekmez, bu da ARM ağ geçitleri için çapraz derleme yaparken kolaylık sağlar.

Tasarımı, diğer dillerdeki callback tabanlı kütüphanelerden farklıdır ve kod yazmadan önce anlaşılmaya değer:

Event loop'u yoklayan bir şey olmadıkça ağda hiçbir şey olmaz. AsyncClient istekleri sınırlı bir kanala girer; event loop bu kanalı boşaltır, paketleri gönderir, gelen paketleri okur ve ping'leri yönetir. poll() bir hata döndürürse bir sonraki çağrı yeniden bağlanır. Bağlantı mantığınızın yaşadığı yer bu tek döngüdür.

Proje Kurulumu

Yeni bir binary crate oluşturun:

cargo new rust-mqtt-demo
cd rust-mqtt-demo

Cargo.toml dosyasını düzenleyin:

[package]
name = "rust-mqtt-demo"
version = "0.1.0"
edition = "2021"

[dependencies]
rumqttc = "0.25"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

Varsayılan use-rustls özelliği, aws-lc-rs kripto sağlayıcısıyla rustls üzerinden TLS'yi etkinleştirir. Bu sağlayıcıyı derlemek için derleme makinesinde bir C derleyicisi ve CMake gerekir; Windows'ta NASM da gerekebilir. Çalışma zamanı makinelerinde ek bir şeye gerek yoktur.

İlk çalıştırma için broker.connectmq.com adresindeki ücretsiz herkese açık MQTT broker kullanın; bu broker boş olmayan ve benzersiz herhangi bir client ID ile anonim istemcileri kabul eder. connectmq/demo/m3c8wd gibi rastgele bir önek seçin; herkese açık alan herkes tarafından okunabildiği için yalnızca test verisi yayınlayın.

Eksiksiz rumqttc Örneği

src/main.rs dosyasının içeriğini değiştirin:

use rumqttc::{AsyncClient, Event, LastWill, MqttOptions, Packet, QoS, Transport};
use serde::Serialize;
use std::env;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time;

#[derive(Serialize)]
struct Reading {
    device: String,
    temperature: f64,
    ts: u64,
}

fn var(key: &str, default: &str) -> String {
    env::var(key).unwrap_or_else(|_| default.to_string())
}

#[tokio::main]
async fn main() {
    let host = var("MQTT_HOST", "broker.connectmq.com");
    let client_id = var("MQTT_CLIENT_ID", "rust-demo-m3c8wd");
    let prefix = var("MQTT_PREFIX", "connectmq/demo/m3c8wd");
    let status_topic = format!("{prefix}/status");
    let command_filter = format!("{prefix}/cmd/#");
    let telemetry_topic = format!("{prefix}/telemetry");

    let mut options = MqttOptions::new(client_id.clone(), host, 8883);
    options.set_keep_alive(Duration::from_secs(30));
    options.set_clean_session(false);
    options.set_last_will(LastWill::new(&status_topic, "offline", QoS::AtLeastOnce, true));
    options.set_transport(Transport::tls_with_default_config());
    if let Ok(user) = env::var("MQTT_USERNAME") {
        options.set_credentials(user, env::var("MQTT_PASSWORD").unwrap_or_default());
    }

    let (client, mut eventloop) = AsyncClient::new(options, 32);

    // Yayıncı görevi: her 5 saniyede bir okuma gönderir.
    let publisher = client.clone();
    let device = client_id.clone();
    tokio::spawn(async move {
        let mut ticker = time::interval(Duration::from_secs(5));
        let mut temperature = 21.0_f64;
        loop {
            ticker.tick().await;
            temperature += 0.1;
            let ts = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
            let reading = Reading { device: device.clone(), temperature, ts };
            let payload = serde_json::to_vec(&reading).expect("serialize");
            if let Err(e) = publisher
                .publish(&telemetry_topic, QoS::AtLeastOnce, false, payload)
                .await
            {
                eprintln!("publish request failed: {e}");
            }
        }
    });

    // Event loop: bağlantıyı yürütür, yeniden bağlanmaları ve gelen mesajları yönetir.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("connected: {:?}", ack.code);
                // Bloklamayan istekler: poll döngüsü içinde client'ı asla await etmeyin.
                if let Err(e) = client.try_subscribe(&command_filter, QoS::AtLeastOnce) {
                    eprintln!("subscribe request failed: {e}");
                }
                if let Err(e) = client.try_publish(&status_topic, QoS::AtLeastOnce, true, "online") {
                    eprintln!("status request failed: {e}");
                }
            }
            Ok(Event::Incoming(Packet::Publish(msg))) => {
                println!("command on {}: {}", msg.topic, String::from_utf8_lossy(&msg.payload));
            }
            Ok(_) => {}
            Err(e) => {
                eprintln!("connection error: {e}; reconnecting in 5 s");
                time::sleep(Duration::from_secs(5)).await;
            }
        }
    }
}

Çalıştırın:

cargo run

Başka bir terminalde izleyin ve bir komut gönderin:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-m3c8wd -t 'connectmq/demo/m3c8wd/#' -v
mosquitto_pub -h broker.connectmq.com -p 1883 -i send-m3c8wd -t 'connectmq/demo/m3c8wd/cmd/fan' -m 'on'

Her beş saniyede bir telemetri, durum topic'inde retained bir online değeri ve Rust programının yazdırdığı komutu görmelisiniz.

rumqttc İstemcisi Nasıl Çalışır?

rustls ile TLS

Transport::tls_with_default_config(), işletim sisteminin kök sertifikalarına güvenen bir rustls yapılandırması oluşturur. Broker, yaygın tüm işletim sistemlerinin güvendiği bir Let's Encrypt sertifikası kullanır; bu nedenle sertifika dosyalarına gerek yoktur. 8883 portu ve TLS taşıma katmanı tüm kurulumdur. Özel bir CA veya istemci sertifikalarına ihtiyacınız varsa kendiniz bir TlsConfiguration oluşturup Transport::tls_with_config metoduna geçirin.

Döngü içinde neden try_subscribe?

En yaygın rumqttc hatası bir kilitlenmedir (deadlock): event loop'u yoklayan aynı görev içinde client.subscribe(...).await çağırmak. İstek kanalı doluysa await yer açılmasını bekler, ancak yer yalnızca poll() çalıştığında açılır ve bu asla gerçekleşmez. Döngü içinde hemen dönen try_subscribe ve try_publish metotlarını kullanın. Diğer görevlerde asenkron sürümler sorunsuzdur ve size doğal bir geri basınç (back-pressure) sağlar.

Yeniden bağlanma ve oturumlar

poll() bir hata döndürdüğünde bağlantı kopmuştur. poll() metodunu yeniden çağırmak tekrar bağlanır; sleep ise ağ kapalıyken sıkı bir döngüyü önler. set_clean_session(false) ile broker, bağlantılar arasında abonelikleri ve kuyruktaki QoS 1 mesajlarını saklar; istemci de onaylanmamış QoS 1 yayınlarını yeniden gönderir. Yine de her ConnAck sonrasında yeniden abone olmak iyi bir alışkanlıktır, çünkü uzun bir kesintiden sonra broker oturumu sonlandırmış olabilir. Herkese açık broker'da oturumlar en fazla 30 dakika sürer.

Last Will

LastWill, istemci bağlantıyı kesmeden ortadan kaybolursa broker'a retained bir offline mesajı yayınlamasını söyler. Her bağlantıdan sonra gönderilen retained online mesajıyla birlikte herhangi bir pano servisin çalışıp çalışmadığını gösterebilir.

Ö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'ler kimlik bilgilerine bağlıdır; bu nedenle ID birebir eşleşmelidir:

export MQTT_HOST=broker.connectmq.com
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_PREFIX=plant1/line2/edge-gateway
cargo run --release

rumqttc ile MQTT 5

rumqttc::v5 modülü aynı API'yi MQTT 5 tipleriyle yansıtır. Mesaj sona erme süresi (message expiry), user properties ve reason code gibi özellikleri ve clean-session bayrağı yerine bir oturum sona erme aralığı (session expiry interval) sunar. Geçiş büyük ölçüde import'ları değiştirmek ve topic için byte kullanan Publish alanlarına uyum sağlamaktan ibarettir. MQTT 5 özellikleri rehberi hangi özelliklerin kullanılmaya değer olduğunu açıklar.

Uç Ağ Geçitleri için Çapraz Derleme

Rust, Raspberry Pi sınıfı kartlar gibi ARM ağ geçitlerinde popülerdir ve rumqttc OpenSSL gerektirmediği için bu ortamlara kolayca taşınır. Asıl engel, C kodu derleyen kripto sağlayıcısıdır. cross gibi araçlar araç zincirini sizin yerinize yönetir: cross build --release --target aarch64-unknown-linux-gnu, 64 bit bir Pi'ye kopyalayabileceğiniz bir binary üretir. Varsayılan TLS yapılandırması başlangıçta platformun kök sertifikalarını yüklediği için hedef sistemin yine de sistem CA paketine ihtiyaç duyduğunu unutmayın.

Üretim Ortamı Kontrol Listesi

1. Her süreç için benzersiz client ID. Aynı ID'ye sahip iki süreç birbirinin bağlantısını döngü halinde keser.
2. Kanalı boyutlandırın. AsyncClient::new metoduna geçirilen kapasite kuyruktaki istekleri sınırlar. Ani yük üreten yayıncılar daha fazla pay gerektirir.
3. Hatalarda geri çekilin. Hatalardan sonra poll() metodunu asla sıkı bir döngüde çağırmayın; tercihen giderek artan gecikmelerle bekleyin.
4. QoS'u bilinçli seçin. Çoğu telemetri için QoS 1 uygundur; Free plan QoS 0 ve 1'i destekler. MQTT QoS seviyeleri açıklaması yazısına bakın.
5. Payload'ları küçük tutun. Herkese açık broker mesaj başına 64 KB'a, Free plan ise 256 KB'a kadar kabul eder.
6. Gizli bilgileri binary içinde değil, ortam değişkenlerinde veya bir gizli bilgi deposunda tutun.

Sorun Giderme

Crate'in API başvurusu docs.rs üzerinde bulunur.

İlgili Eğitimler

Sisteminizin bazı bölümleri başka dillerde yazılmışsa Go MQTT istemcisi eğitimi ve Java MQTT istemcisi rehberi aynı telemetri ve komut desenini uygular. MQTT istemci kütüphanelerine genel bakış farklı ekosistemlerdeki istemcileri karşılaştırır.

Rust Servislerinizi ConnectMQ Üzerinde Çalıştırın

Tam yönetilen bir MQTT hizmeti, Rust ağ geçidinizin yalnızca tek bir giden TLS bağlantısına ihtiyaç duyması demektir; yama yapılacak bir broker ya da yenilenecek sertifika yoktur. ConnectMQ, 8883 portunda TLS ile MQTT 3.1.1 ve 5.0'ı destekler ve istemcinizin ne gönderdiğini kontrol etmeniz için canlı bir topic gezgini sunar. app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun ve yukarıdaki dört ortam değişkenini ayarlayın.

Sık Sorulan Sorular

rumqttc TLS'yi destekliyor mu?

Evet. Varsayılan use-rustls özelliğiyle Transport::tls_with_default_config(), işletim sisteminin güvenilir kök sertifikalarını kullanarak TLS üzerinden bağlanır. 8883 portunu kullanın.

rumqttc otomatik olarak yeniden bağlanır mı?

Bir hatadan sonra eventloop.poll() metodunu tekrar çağırdığınızda yeniden bağlanır. Hatalardan sonra kısa bir beklemeyle döngü içinde yoklamaya devam edin ve her ConnAck sonrasında yeniden abone olun.

rumqttc MQTT 5'i destekliyor mu?

Evet; MQTT 5 özelliklerini, reason code'ları ve oturum sona erme süresini sağlayan rumqttc::v5 modülü aracılığıyla.

rumqttc programım neden donuyor?

Genellikle event loop'u yoklayan görev içinde subscribe veya publish await edildiği içindir. Orada bunların yerine try_subscribe ve try_publish kullanın.