Rust MQTT mit rumqttc: Asynchrones Publish/Subscribe über TLS mit Tokio

Dieses Rust-MQTT-rumqttc-Tutorial zeigt, wie Ihnen rumqttc einen reinen Rust-Client mit Async-Unterstützung bietet, der sich ganz natürlich in Tokio-Dienste einfügt, vom Edge-Gateway bis zum Backend-Ingest-Worker. Sie bauen einen vollständigen rumqttc-Client, der sich per TLS mit rustls auf Port 8883 verbindet, JSON-Telemetrie veröffentlicht, Befehle empfängt und Netzwerkabbrüche übersteht.

Warum rumqttc für Rust und MQTT

rumqttc ist die Client-Hälfte des rumqtt-Projekts. Es ist vollständig in Rust geschrieben, unterstützt MQTT 3.1.1 im Crate-Root und MQTT 5 im Modul v5 und bietet sowohl einen synchronen Client als auch einen auf Tokio basierenden AsyncClient. TLS kommt standardmäßig von rustls, Sie brauchen also kein OpenSSL auf dem Zielsystem, was beim Cross-Compiling für ARM-Gateways praktisch ist.

Das Design unterscheidet sich von callback-basierten Bibliotheken in anderen Sprachen, und es lohnt sich, es vor dem Programmieren zu verstehen:

Im Netzwerk passiert nichts, solange niemand den Event Loop pollt. Anfragen von AsyncClient landen in einem begrenzten Channel; der Event Loop leert ihn, sendet Pakete, liest eingehende Pakete und kümmert sich um Pings. Gibt poll() einen Fehler zurück, verbindet sich der nächste Aufruf neu. In dieser einen Schleife liegt Ihre gesamte Verbindungslogik.

Projekt einrichten

Legen Sie ein neues Binary-Crate an:

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

Bearbeiten Sie Cargo.toml:

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

Das Standard-Feature use-rustls aktiviert TLS über rustls mit dem Kryptoprovider aws-lc-rs. Zum Bauen dieses Providers brauchen Sie einen C-Compiler und CMake auf dem Build-Rechner; unter Windows kann zusätzlich NASM nötig sein. Die Zielsysteme zur Laufzeit brauchen nichts weiter.

Für den ersten Lauf verwenden Sie den kostenlosen öffentlichen MQTT-Broker unter broker.connectmq.com, der anonyme Clients mit jeder nicht leeren, eindeutigen Client ID akzeptiert. Wählen Sie ein zufälliges Präfix wie connectmq/demo/m3c8wd; der öffentliche MQTT-Broker ist für jeden lesbar und für Tests und Prototypen gedacht, veröffentlichen Sie dort also nur Testdaten und keine Produktionsdaten.

Das vollständige rumqttc-Beispiel

Ersetzen Sie src/main.rs:

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

    // Publisher-Task: sendet alle 5 Sekunden einen Messwert.
    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: treibt die Verbindung, behandelt Reconnects und eingehende Nachrichten.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("connected: {:?}", ack.code);
                // Nicht blockierende Anfragen: niemals den Client innerhalb der Poll-Schleife awaiten.
                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;
            }
        }
    }
}

Starten Sie es:

cargo run

Beobachten Sie in einem anderen Terminal und senden Sie einen Befehl:

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'

Sie sollten alle fünf Sekunden Telemetrie sehen, ein retained online auf dem Status-Topic und den vom Rust-Programm ausgegebenen Befehl.

Wie der rumqttc-Client funktioniert

TLS mit rustls

Transport::tls_with_default_config() erstellt eine rustls-Konfiguration, die den Root-Zertifikaten des Betriebssystems vertraut. Der Broker verwendet ein Let's-Encrypt-Zertifikat, dem jedes gängige Betriebssystem vertraut, daher sind keine Zertifikatsdateien nötig. Port 8883 plus TLS-Transport ist das gesamte Setup. Brauchen Sie eine eigene CA oder Client-Zertifikate, erstellen Sie selbst eine TlsConfiguration und übergeben sie an Transport::tls_with_config.

Warum try_subscribe innerhalb der Schleife

Der häufigste rumqttc-Fehler ist ein Deadlock: client.subscribe(...).await im selben Task aufzurufen, der den Event Loop pollt. Ist der Anfrage-Channel voll, wartet das Await auf freien Platz, doch Platz wird nur frei, wenn poll() läuft, und das geschieht dann nie. Verwenden Sie innerhalb der Schleife try_subscribe und try_publish, die sofort zurückkehren. In anderen Tasks sind die Async-Varianten in Ordnung und liefern Ihnen natürlichen Back-Pressure.

Reconnects und Sessions

Gibt poll() einen Fehler zurück, ist die Verbindung weg. Ein erneuter Aufruf von poll() verbindet neu, und das Sleep verhindert eine enge Schleife, solange das Netzwerk ausgefallen ist. Mit set_clean_session(false) hält der Broker Abonnements und gepufferte QoS-1-Nachrichten zwischen Verbindungen vor, und der Client sendet unbestätigte QoS-1-Publishes erneut. Bei jedem ConnAck neu zu abonnieren ist trotzdem eine gute Gewohnheit, weil der Broker die Session nach einem langen Ausfall verworfen haben kann. Auf dem öffentlichen MQTT-Broker halten Sessions höchstens 30 Minuten.

Last Will

LastWill weist den Broker an, ein retained offline zu veröffentlichen, wenn der Client ohne Trennen verschwindet. Zusammen mit dem retained online, das nach jeder Verbindung gesendet wird, kann jedes Dashboard anzeigen, ob der Dienst läuft.

Umstieg auf einen privaten Broker

Für den Produktivbetrieb legen Sie im ConnectMQ-Portal einen Broker an und verwenden den generierten Benutzernamen, das Passwort und die Client ID. Client IDs sind an die Zugangsdaten gebunden, die ID muss also exakt übereinstimmen:

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

Läuft der Client als Edge-Gateway in einem Werk, lohnt es sich, das Präfix gleich an einem Unified Namespace (UNS) auszurichten: Die ISA-95-Hierarchie (Enterprise → Site → Area → Line → Cell) wird zur Topic-Struktur, etwa acme/werk-stuttgart/montage/linie-2/edge-gateway. So landen die Daten des Rust-Dienstes an derselben Stelle, an der MES, SCADA und Analysewerkzeuge sie erwarten. Hintergründe dazu finden Sie im Beitrag Sparkplug B und Unified Namespace.

MQTT 5 mit rumqttc

Das Modul rumqttc::v5 spiegelt dieselbe API mit MQTT-5-Typen. Es bietet Properties wie Message-Expiry, User Properties und Reason Codes sowie ein Session-Expiry-Intervall anstelle des Clean-Session-Flags. Der Wechsel besteht hauptsächlich darin, Imports anzupassen und mit Publish-Feldern umzugehen, die Bytes für Topics verwenden. Der Leitfaden zu den MQTT-5-Funktionen erklärt, welche Properties sich lohnen.

Cross-Compiling für Edge-Gateways

Rust ist auf ARM-Gateways wie Boards der Raspberry-Pi-Klasse und auf Industrie-PCs neben der SPS beliebt, und rumqttc lässt sich dorthin gut mitnehmen, weil es kein OpenSSL braucht. Die größte Hürde ist der Kryptoprovider, der C-Code kompiliert. Werkzeuge wie cross übernehmen die Toolchain für Sie: cross build --release --target aarch64-unknown-linux-gnu erzeugt ein Binary, das Sie auf einen 64-Bit-Pi kopieren können. Denken Sie daran, dass das Zielsystem trotzdem das CA-Bundle des Systems braucht, weil die Standard-TLS-Konfiguration beim Start die Root-Zertifikate der Plattform lädt.

Produktions-Checkliste

1. Eindeutige Client ID pro Prozess. Zwei Prozesse mit derselben ID trennen sich gegenseitig in einer Endlosschleife.
2. Den Channel dimensionieren. Die an AsyncClient::new übergebene Kapazität begrenzt die Anfragen in der Warteschlange. Publisher mit Lastspitzen brauchen mehr Spielraum.
3. Bei Fehlern zurückhalten. Rufen Sie poll() nach Fehlern nie in einer engen Schleife auf; schlafen Sie, idealerweise mit wachsenden Verzögerungen.
4. QoS bewusst wählen. QoS 1 eignet sich für die meiste Telemetrie; der Free-Tarif unterstützt QoS 0 und 1. Siehe MQTT-QoS-Stufen erklärt.
5. Payloads klein halten. Der öffentliche MQTT-Broker akzeptiert bis zu 64 KB pro Nachricht, der Free-Tarif 256 KB.
6. Geheimnisse in der Umgebung halten oder in einem Secret Store, nicht im Binary.

Fehlersuche

Die API-Referenz des Crates finden Sie auf docs.rs.

Verwandte Tutorials

Sind Teile Ihres Systems in anderen Sprachen geschrieben, setzen das Go-MQTT-Client-Tutorial und der Leitfaden zum Java-MQTT-Client dasselbe Muster aus Telemetrie und Befehlen um. Der Überblick über MQTT-Client-Bibliotheken vergleicht Clients über Ökosysteme hinweg.

Betreiben Sie Ihre Rust-Dienste mit ConnectMQ

Ein verwalteter MQTT-Broker bedeutet, dass Ihr Rust-Gateway nur eine ausgehende TLS-Verbindung braucht, ohne Broker, den Sie patchen, oder Zertifikate, die Sie erneuern müssen. ConnectMQ unterstützt MQTT 3.1.1 und 5.0 mit TLS auf 8883 und einem Live-Topic-Explorer, mit dem Sie prüfen, was Ihr Client sendet. Erstellen Sie einen kostenlosen Serverless-Broker unter app.connectmq.com ohne Kreditkarte und setzen Sie die vier obigen Umgebungsvariablen.

Häufig gestellte Fragen

Unterstützt rumqttc TLS?

Ja. Mit dem Standard-Feature use-rustls verbindet sich Transport::tls_with_default_config() per TLS und nutzt die vertrauenswürdigen Root-Zertifikate des Betriebssystems. Verwenden Sie Port 8883.

Verbindet sich rumqttc automatisch neu?

Es verbindet sich neu, wenn Sie nach einem Fehler erneut eventloop.poll() aufrufen. Pollen Sie weiter in einer Schleife mit kurzem Sleep nach Fehlern und abonnieren Sie bei jedem ConnAck neu.

Unterstützt rumqttc MQTT 5?

Ja, über das Modul rumqttc::v5, das MQTT-5-Properties, Reason Codes und Session-Expiry bereitstellt.

Warum friert mein rumqttc-Programm ein?

Meist, weil es subscribe oder publish in dem Task awaitet, der den Event Loop pollt. Verwenden Sie dort stattdessen try_subscribe und try_publish.