MQTT in C# mit MQTTnet 4 und 5: Verbinden, Abonnieren, Reconnect

MQTTnet ist die Standardbibliothek für MQTT unter .NET. Für einen Client in C# erzeugst du einen IMqttClient aus der Factory, beschreibst die Verbindung mit MqttClientOptionsBuilder, hängst einen ApplicationMessageReceivedAsync-Handler an und rufst ConnectAsync, SubscribeAsync und PublishAsync auf. Seit MQTTnet 5 gibt es keinen Managed Client mehr – den Reconnect schreibst du selbst, das sind rund zwanzig Zeilen.

Dieses Tutorial nutzt MQTTnet 5 und markiert jede Stelle, an der Version 4 abweicht, sodass du mit beiden Versionen folgen kannst. Alle Beispiele verbinden sich mit broker.connectmq.com und nutzen Topics unter testmqtt/dotnet/.

Installieren und Version wählen

dotnet new console -n MqttDemo && cd MqttDemo
dotnet add package MQTTnet

Damit landet die neueste Major-Version im Projekt. Pflegst du eine bestehende MQTTnet-4-Codebasis, pinne sie mit --version 4.*, bis du migrieren willst. Die Unterschiede, auf die du zuerst triffst:

ConvertPayloadToString() funktioniert in beiden Versionen, deshalb verwenden es die Beispiele.

Verbinden

using MQTTnet;            // v4: zusätzlich `using MQTTnet.Client;`
using MQTTnet.Formatter;
using MQTTnet.Protocol;

var factory = new MqttClientFactory();   // v4: new MqttFactory()
using var client = factory.CreateMqttClient();

var options = new MqttClientOptionsBuilder()
    .WithTcpServer("broker.connectmq.com", 1883)
    .WithClientId($"dotnet-{Guid.NewGuid():N}"[..20])
    .WithCredentials(Environment.GetEnvironmentVariable("MQTT_USER"),
                     Environment.GetEnvironmentVariable("MQTT_PASS"))
    .WithProtocolVersion(MqttProtocolVersion.V500)
    .WithKeepAlivePeriod(TimeSpan.FromSeconds(30))
    .WithCleanStart()
    .Build();

var result = await client.ConnectAsync(options, CancellationToken.None);
Console.WriteLine($"Verbunden: {result.ResultCode}");

Auf dem geteilten Test-Endpoint kannst du WithCredentials weglassen; bei deinem eigenen Broker übergibst du Benutzername und Passwort aus Umgebungsvariablen statt sie in den Code zu schreiben. Lehnt der Broker die Verbindung ab, wirft ConnectAsync eine MqttClientConnectingFailedException; ihr ResultCode (etwa BadUserNameOrPassword oder NotAuthorized) verrät den Grund. Die Referenz der MQTT Reason Codes listet alle Werte, die häufigsten Exceptions erklärt MQTTnet-Fehler.

Last Will hinzufügen

Die meisten Clients sollten melden, wenn sie verschwinden. Ergänze denselben Options-Builder um eine Will-Nachricht:

    .WithWillTopic("testmqtt/dotnet/status")
    .WithWillPayload("offline")
    .WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
    .WithWillRetain()

Der Broker veröffentlicht den Will nur, wenn die Verbindung ohne DISCONNECT-Paket endet: Absturz, Netzverlust oder Keep-Alive-Timeout. Ein sauberes DisconnectAsync() unterdrückt ihn – veröffentliche also nach dem Verbinden selbst einen retained Status online und vor einem geplanten Shutdown offline. Mit MQTT 5 gibt .WithWillDelayInterval(10) dem Client ein paar Sekunden zum Wiederkommen, bevor der Will rausgeht; das verhindert einen flackernden Status bei kurzen Netzaussetzern. Details findest du unter MQTT Last Will.

Abonnieren und Nachrichten verarbeiten

Registriere den Handler vor dem Abonnieren, sonst können Retained Messages, die direkt nach dem SUBACK kommen, eintreffen, bevor jemand zuhört:

client.ApplicationMessageReceivedAsync += e =>
{
    var msg = e.ApplicationMessage;
    Console.WriteLine($"{msg.Topic} (QoS {(int)msg.QualityOfServiceLevel}): {msg.ConvertPayloadToString()}");
    return Task.CompletedTask;
};

var subscribe = factory.CreateSubscribeOptionsBuilder()
    .WithTopicFilter(f => f.WithTopic("testmqtt/dotnet/sensors/#")
                           .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
    .Build();

var subResult = await client.SubscribeAsync(subscribe, CancellationToken.None);
foreach (var item in subResult.Items)
    Console.WriteLine($"{item.TopicFilter.Topic}: {item.ResultCode}");

Gib das Subscribe-Ergebnis immer aus. Ein Broker, der eine Subscription per ACL verweigert, schließt die Verbindung nicht, sondern liefert im SUBACK einen Fehlercode wie NotAuthorized – und dein Handler feuert nie. Ein Blick auf ResultCode macht aus einem stillen Fehler eine Logzeile.

Der Handler läuft im internen Dispatcher von MQTTnet. Halte ihn kurz: parsen, Arbeit abgeben (an einen Channel<T>, eine Queue oder einen Hintergrund-Task) und zurückkehren. Lange Arbeit im Handler verzögert die Acknowledgements und jede nachfolgende Nachricht.

Veröffentlichen

using System.Text.Json;

var payload = JsonSerializer.Serialize(new { celsius = 21.5, ts = DateTimeOffset.UtcNow });

var message = new MqttApplicationMessageBuilder()
    .WithTopic("testmqtt/dotnet/sensors/lab")
    .WithPayload(payload)
    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
    .WithRetainFlag(false)
    .WithContentType("application/json")      // nur MQTT 5
    .WithUserProperty("source", "mqttnet-demo") // nur MQTT 5
    .Build();

var pub = await client.PublishAsync(message, CancellationToken.None);
Console.WriteLine($"Publish: {pub.ReasonCode}");

Bei QoS 1 ist PublishAsync fertig, wenn das PUBACK eintrifft – der Reason Code ist also die Antwort des Brokers. Bei QoS 0 ist der Aufruf fertig, sobald das Paket in den Socket geschrieben ist. Große Payloads kann der Broker wegen seiner maximalen Paketgröße ablehnen; der Paketgrößen-Rechner zeigt, wie groß eine Nachricht auf der Leitung wirklich ist.

TLS und WebSockets

Tausche den Transport im Options-Builder aus, alles andere bleibt gleich:

var tlsOptions = new MqttClientOptionsBuilder()
    .WithTcpServer("broker.connectmq.com", 8883)
    .WithTlsOptions(o => o.UseTls())
    .WithCredentials(Environment.GetEnvironmentVariable("MQTT_USER"),
                     Environment.GetEnvironmentVariable("MQTT_PASS"))
    .Build();

var wssOptions = new MqttClientOptionsBuilder()
    .WithWebSocketServer(o => o.WithUri("wss://broker.connectmq.com:8084/mqtt"))
    .WithTlsOptions(o => o.UseTls())
    .Build();

Das Broker-Zertifikat wird gegen den Trust Store des Betriebssystems geprüft. Widerstehe der Versuchung, WithCertificateValidationHandler(_ => true) einzubauen, nur damit ein Handshake-Fehler verschwindet; such lieber die Ursache, am besten zuerst mit einem Test der TLS-Verbindung im Terminal. Ältere MQTTnet-4-Versionen nutzten .WithTls(), das inzwischen obsolet ist.

Reconnect ohne ManagedMqttClient

MQTTnet 4 bot den ManagedMqttClient, der für dich neu verband und Nachrichten zwischenspeicherte. In Version 5 ist er weg, und die offiziellen Beispiele verwenden stattdessen eine einfache Schleife. Hier eine vollständige Version:

client.ConnectedAsync += async e =>
{
    // Mit Clean Start hat der Broker unsere Subscriptions vergessen; mit persistenter Session vielleicht nicht.
    if (!e.ConnectResult.IsSessionPresent)
        await client.SubscribeAsync(subscribe, CancellationToken.None);
    Console.WriteLine("Verbunden, Session vorhanden: " + e.ConnectResult.IsSessionPresent);
};

client.DisconnectedAsync += e =>
{
    Console.WriteLine($"Getrennt ({e.Reason}): {e.Exception?.Message}");
    return Task.CompletedTask;
};

using var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, a) => { a.Cancel = true; cts.Cancel(); };

var delay = TimeSpan.FromSeconds(1);
while (!cts.IsCancellationRequested)
{
    try
    {
        if (!await client.TryPingAsync(cts.Token))
        {
            await client.ConnectAsync(options, cts.Token);
            delay = TimeSpan.FromSeconds(1);           // Backoff nach Erfolg zurücksetzen
        }
    }
    catch (Exception ex) when (ex is not OperationCanceledException)
    {
        Console.WriteLine($"Reconnect fehlgeschlagen: {ex.Message}");
        delay = TimeSpan.FromSeconds(Math.Min(delay.TotalSeconds * 2, 60));
    }

    try { await Task.Delay(delay + TimeSpan.FromMilliseconds(Random.Shared.Next(500)), cts.Token); }
    catch (OperationCanceledException) { break; }
}

await client.DisconnectAsync();

Drei Details machen das robust. Die Subscription steckt in ConnectedAsync und wird so nach jedem Reconnect wiederhergestellt, nicht nur beim ersten. Die Wartezeit wächst exponentiell mit etwas zufälligem Jitter, damit tausend Clients, die gleichzeitig den Broker verlieren, nicht alle in derselben Sekunde neu verbinden – warum das wichtig ist, erklären die Best Practices für Reconnects. Und TryPingAsync dient zugleich als Health Check: Es liefert false, wenn der Client nicht verbunden ist, aber auch bei einer halboffenen Verbindung.

Puffern während der Trennung

Der Managed Client hat auch ausgehende Nachrichten gepuffert. Brauchst du das, setze einen Channel<MqttApplicationMessage> zwischen deinen Code und den Client: Produzenten schreiben in den Channel, ein einzelner Consumer-Task liest daraus und ruft PublishAsync nur auf, wenn client.IsConnected true ist, und wiederholt eine Nachricht, falls das Publish eine Exception wirft. Begrenze den Channel (Channel.CreateBounded), damit ein langer Ausfall nicht den Speicher sprengt, und entscheide bewusst, ob bei vollem Channel die ältesten oder die neuesten Nachrichten verworfen werden.

MQTT-5-Funktionen, die sich lohnen

Sobald WithProtocolVersion(MqttProtocolVersion.V500) gesetzt ist, zahlen sich einige Optionen sofort aus:

• Session Expiry – .WithSessionExpiryInterval(3600) zusammen mit .WithCleanStart(false) hält Subscriptions und eingereihte QoS-1-Nachrichten eine Stunde nach einer Trennung vor. Teste das gegen einen privaten Broker; geteilte Test-Endpoints halten in der Regel keine Offline-Sessions vor.
• Reason Codes und Reason Strings – Connect-, Subscribe- und Publish-Ergebnisse enthalten ResultCode/ReasonCode und oft einen ReasonString. Logge sie; sie machen aus „geht nicht“ ein „Quota überschritten“.
• User Properties – Key/Value-Metadaten wie eine Trace-ID oder die Firmware-Version, ohne das Payload-Format zu ändern.
• Message Expiry – .WithMessageExpiryInterval(60) an einer Nachricht verhindert, dass veraltete Befehle an ein Gerät gehen, das sich eine Stunde später wieder verbindet.
• Request/Response – .WithResponseTopic() und .WithCorrelationData() sind der Standardweg für RPC über MQTT.

Testen

Starte die Konsolen-App, öffne den Online-MQTT-Client und abonniere testmqtt/dotnet/#. Der JSON-Messwert sollte auf testmqtt/dotnet/sensors/lab ankommen. Veröffentliche aus dem Browser auf testmqtt/dotnet/sensors/test, und der C#-Handler gibt die Nachricht aus. Um die Reconnect-Schleife zu testen, trenne dein Netzwerk für ein paar Sekunden und beobachte, wie das Log von Getrennt zu Verbunden, Session vorhanden: False und einem frischen SUBACK wechselt.

Nachrichten deines MQTTnet-Clients live ansehen

Baust du eine Web-API oder einen Worker Service, ist der nächste Schritt, diesen Client in einen Hosted Service mit Dependency Injection und Health Checks zu verlagern – siehe MQTT in ASP.NET Core.

MQTTnet bietet eine kleine, explizite API: einen Options-Builder, drei asynchrone Aufrufe und eine Handvoll Events. Ergänze eine Reconnect-Schleife, die in ConnectedAsync neu abonniert, logge jeden Result Code, und derselbe Client funktioniert vom Konsolentest auf dem geteilten Endpoint bis zum Produktionsdienst auf deinem eigenen Broker.

Häufig gestellte Fragen