MQTT en C# avec MQTTnet 4 et 5 : connexion, abonnement, reconnexion

MQTTnet est la bibliothèque MQTT de référence pour .NET. Pour créer un client en C#, obtenez un IMqttClient depuis la factory, décrivez la connexion avec MqttClientOptionsBuilder, branchez un handler ApplicationMessageReceivedAsync, puis appelez ConnectAsync, SubscribeAsync et PublishAsync. Depuis MQTTnet 5, il n'y a plus de managed client : la reconnexion est à écrire vous-même, en une vingtaine de lignes.

Ce tutoriel utilise MQTTnet 5 et signale chaque endroit où la version 4 diffère, pour que vous puissiez le suivre avec l'une ou l'autre. Tous les exemples se connectent à broker.connectmq.com et utilisent des topics sous testmqtt/dotnet/.

Installer et choisir une version

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

Cela installe la dernière version majeure. Si vous maintenez une base de code MQTTnet 4, figez-la avec --version 4.* jusqu'à ce que vous soyez prêt à migrer. Les différences que vous rencontrerez en premier :

ConvertPayloadToString() fonctionne dans les deux versions, c'est pourquoi les exemples l'utilisent.

Se connecter

using MQTTnet;            // v4 : ajoutez aussi `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($"Connecté : {result.ResultCode}");

Sur l'endpoint de test partagé, vous pouvez omettre WithCredentials ; sur votre propre broker, passez l'identifiant et le mot de passe via des variables d'environnement plutôt que de les écrire en dur. Si le broker refuse la connexion, ConnectAsync lève une MqttClientConnectingFailedException ; son ResultCode (par exemple BadUserNameOrPassword ou NotAuthorized) en donne la raison. La référence des reason codes MQTT liste toutes les valeurs, et les exceptions les plus courantes sont expliquées dans erreurs MQTTnet.

Ajouter un Last Will

La plupart des clients devraient signaler leur disparition. Ajoutez un message will au même builder d'options :

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

Le broker ne publie le will que si la connexion se termine sans paquet DISCONNECT : crash, perte réseau ou timeout de keep-alive. Un DisconnectAsync() propre le supprime ; publiez donc vous-même un statut retenu online après la connexion et offline avant un arrêt planifié. Avec MQTT 5, .WithWillDelayInterval(10) laisse au client quelques secondes pour revenir avant l'envoi du will, ce qui évite un statut qui clignote lors de brèves coupures réseau. MQTT Last Will entre dans les détails.

S'abonner et traiter les messages

Enregistrez le handler avant de vous abonner ; sinon, les messages retenus livrés juste après le SUBACK peuvent arriver avant que quiconque n'écoute :

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

Affichez toujours le résultat du subscribe. Un broker qui refuse un abonnement via son ACL ne ferme pas la connexion : il renvoie simplement un code d'échec comme NotAuthorized dans le SUBACK, et votre handler ne se déclenche jamais. Vérifier ResultCode transforme un échec silencieux en ligne de log.

Le handler s'exécute dans le dispatcher interne de MQTTnet. Gardez-le court : parsez, déléguez le travail (à un Channel<T>, une file ou une tâche de fond) et rendez la main. Un traitement long dans le handler retarde les acquittements et tous les messages suivants.

Publier

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")      // MQTT 5 uniquement
    .WithUserProperty("source", "mqttnet-demo") // MQTT 5 uniquement
    .Build();

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

En QoS 1, PublishAsync se termine à l'arrivée du PUBACK : le reason code est donc la réponse du broker. En QoS 0, il se termine dès que le paquet est écrit dans le socket. Le broker peut rejeter les gros payloads à cause de sa taille maximale de paquet ; le calculateur de taille de paquet montre la taille réelle d'un message sur le réseau.

TLS et WebSockets

Changez le transport dans le builder d'options ; tout le reste est identique :

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

Le certificat du broker est validé via le magasin de confiance du système. Résistez à la tentation d'ajouter WithCertificateValidationHandler(_ => true) pour faire disparaître une erreur de handshake ; cherchez plutôt la cause, en commençant par tester la connexion TLS depuis un terminal. Les anciennes versions de MQTTnet 4 utilisaient .WithTls(), désormais obsolète.

Se reconnecter sans ManagedMqttClient

MQTTnet 4 proposait ManagedMqttClient, qui se reconnectait et mettait les messages en file pour vous. Il a disparu en version 5, et les exemples officiels utilisent à la place une simple boucle. En voici une version complète :

client.ConnectedAsync += async e =>
{
    // Avec clean start, le broker a oublié nos abonnements ; avec une session persistante, peut-être pas.
    if (!e.ConnectResult.IsSessionPresent)
        await client.SubscribeAsync(subscribe, CancellationToken.None);
    Console.WriteLine("Connecté, session présente : " + e.ConnectResult.IsSessionPresent);
};

client.DisconnectedAsync += e =>
{
    Console.WriteLine($"Déconnecté ({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);           // remet le backoff à zéro après succès
        }
    }
    catch (Exception ex) when (ex is not OperationCanceledException)
    {
        Console.WriteLine($"Échec de reconnexion : {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();

Trois détails la rendent robuste. L'abonnement se trouve dans ConnectedAsync : il est donc restauré après chaque reconnexion, pas seulement la première. Le délai croît de façon exponentielle avec un peu de jitter aléatoire, pour que mille clients qui perdent le broker en même temps ne se reconnectent pas tous dans la même seconde — les bonnes pratiques de reconnexion expliquent pourquoi c'est important. Enfin, TryPingAsync sert aussi de health check : il renvoie false aussi bien quand le client n'est pas connecté que lorsque la connexion est à moitié ouverte.

Tampon hors ligne

Le managed client mettait aussi en file les messages sortants. Si vous en avez besoin, placez un Channel<MqttApplicationMessage> entre votre code et le client : les producteurs écrivent dans le canal, et une seule tâche consommatrice le lit et appelle PublishAsync uniquement quand client.IsConnected vaut true, en réessayant un message si la publication lève une exception. Bornez le canal (Channel.CreateBounded) pour qu'une longue coupure n'épuise pas la mémoire, et décidez explicitement s'il faut jeter les messages les plus anciens ou les plus récents quand il est plein.

Fonctionnalités MQTT 5 utiles

Une fois WithProtocolVersion(MqttProtocolVersion.V500) défini, quelques options sont rentables tout de suite :

• Session expiry — .WithSessionExpiryInterval(3600) combiné à .WithCleanStart(false) conserve abonnements et messages QoS 1 en file pendant une heure après une déconnexion. Testez-le sur un broker privé ; les endpoints de test partagés ne conservent généralement pas les sessions hors ligne.
• Reason codes et reason strings — les résultats de connect, subscribe et publish contiennent ResultCode/ReasonCode et souvent un ReasonString. Journalisez-les : « ça ne marche pas » devient « quota dépassé ».
• User properties — des métadonnées clé/valeur comme un trace ID ou une version de firmware, sans changer le format du payload.
• Message expiry — .WithMessageExpiryInterval(60) sur un message empêche la livraison de commandes périmées à un appareil qui se reconnecte une heure plus tard.
• Request/response — .WithResponseTopic() et .WithCorrelationData() offrent une manière standard d'implémenter du RPC sur MQTT.

Tester

Lancez l'application console, ouvrez le client MQTT en ligne et abonnez-vous à testmqtt/dotnet/#. La mesure JSON doit arriver sur testmqtt/dotnet/sensors/lab. Publiez depuis le navigateur sur testmqtt/dotnet/sensors/test et le handler C# l'affiche. Pour tester la boucle de reconnexion, coupez le réseau quelques secondes et regardez le log passer de Déconnecté à Connecté, session présente : False, suivi d'un nouveau SUBACK.

Voir en direct les messages de votre client MQTTnet

Si vous construisez une web API ou un worker service, l'étape suivante consiste à déplacer ce client dans un hosted service avec injection de dépendances et health checks — voir MQTT dans ASP.NET Core.

MQTTnet offre une API petite et explicite : un builder d'options, trois appels async et quelques événements. Ajoutez une boucle de reconnexion qui se réabonne dans ConnectedAsync, journalisez chaque code de résultat, et le même client fonctionne d'un test console sur l'endpoint partagé jusqu'à un service de production sur votre propre broker.

Questions fréquentes