MQTT dans ASP.NET Core : BackgroundService et health checks

Pour consommer MQTT dans ASP.NET Core, enregistrez un seul IMqttClient MQTTnet dans l'injection de dépendances, connectez-le et reconnectez-le depuis un BackgroundService, poussez les messages entrants dans un Channel<T> et traitez-les dans un second hosted service qui crée un scope DI par message. Un health check expose l'état de la connexion, et vos endpoints d'API publient via le même client.

Ce tutoriel construit ce service pas à pas avec MQTTnet 5 sur .NET 8 ou plus récent. Si vous n'avez jamais utilisé MQTTnet, lisez d'abord MQTT en C# avec MQTTnet ; cet article se concentre sur l'hébergement.

L'architecture

Composant	Durée de vie	Rôle
`IMqttClient`	Singleton	Une connexion par instance de l'app, partagée par tout le monde
`MqttConnectionService`	Hosted service	Connecte, reconnecte, s'abonne, se déconnecte à l'arrêt
`Channel<InboundMessage>`	Singleton	Tampon borné entre le réseau et votre logique métier
`TelemetryConsumer`	Hosted service	Lit le canal, crée un scope, appelle votre handler
`ITelemetryHandler`	Scoped	Votre code — peut utiliser `DbContext` et d'autres services scoped
`MqttHealthCheck`	Transient	Indique si le client est connecté

Séparer réception et traitement est important. Le handler d'événements MQTTnet s'exécute dans le dispatcher du client ; s'il attend une base de données, tous les messages suivants attendent aussi, et la gestion du keep-alive peut en pâtir. Le canal découple les deux et fournit un backpressure naturel.

Projet et configuration

Créer le projet

dotnet new web -n MqttWorker && cd MqttWorker
dotnet add package MQTTnet
dotnet user-secrets init
dotnet user-secrets set "Mqtt:Username" "username"
dotnet user-secrets set "Mqtt:Password" "password"

appsettings.json

{
  "Mqtt": {
    "Host": "broker.connectmq.com",
    "Port": 8883,
    "UseTls": true,
    "Topics": [ "testmqtt/aspnet/telemetry/#" ]
  }
}

Les identifiants vivent dans les user secrets en local et dans des variables d'environnement (Mqtt__Username, Mqtt__Password) ou un coffre de secrets en production — jamais dans appsettings.json. Liez la section à une classe de configuration :

MqttSettings.cs

public sealed class MqttSettings
{
    public string Host { get; set; } = "broker.connectmq.com";
    public int Port { get; set; } = 8883;
    public bool UseTls { get; set; } = true;
    public string? Username { get; set; }
    public string? Password { get; set; }
    public string ClientId { get; set; } = $"aspnet-{Environment.MachineName}";
    public string[] Topics { get; set; } = [];
}

public sealed record InboundMessage(string Topic, string Payload, DateTimeOffset ReceivedAt);

Le client ID par défaut utilise le nom de la machine, c'est-à-dire le nom du pod dans Kubernetes, donc unique par réplica. Deux instances avec le même client ID se déconnecteraient mutuellement du broker en boucle.

Enregistrer les services

Program.cs

using System.Text.Json;
using System.Text.RegularExpressions;
using System.Threading.Channels;
using Microsoft.AspNetCore.Diagnostics.HealthChecks;
using Microsoft.Extensions.Diagnostics.HealthChecks;
using MQTTnet;
using MQTTnet.Protocol;

var builder = WebApplication.CreateBuilder(args);

builder.Services.Configure<MqttSettings>(builder.Configuration.GetSection("Mqtt"));
builder.Services.AddSingleton<IMqttClient>(_ => new MqttClientFactory().CreateMqttClient());
builder.Services.AddSingleton(_ => Channel.CreateBounded<InboundMessage>(
    new BoundedChannelOptions(1_000) { FullMode = BoundedChannelFullMode.Wait, SingleReader = true }));

builder.Services.AddHostedService<MqttConnectionService>();
builder.Services.AddHostedService<TelemetryConsumer>();
builder.Services.AddScoped<ITelemetryHandler, TelemetryHandler>();

builder.Services.AddHealthChecks()
    .AddCheck<MqttHealthCheck>("mqtt", failureStatus: HealthStatus.Degraded, tags: ["ready"]);

var app = builder.Build();
app.MapHealthChecks("/healthz/live", new HealthCheckOptions { Predicate = _ => false });
app.MapHealthChecks("/healthz/ready", new HealthCheckOptions { Predicate = c => c.Tags.Contains("ready") });
MapCommandEndpoint(app); // fonction locale, voir plus bas
app.Run();

Avec MQTTnet 4, remplacez MqttClientFactory par MqttFactory et ajoutez using MQTTnet.Client; — le reste du code est identique.

Le service de connexion

Ce hosted service possède la connexion. Il reprend la boucle de reconnexion du tutoriel MQTTnet et se réabonne dans ConnectedAsync, si bien que les abonnements survivent à chaque reconnexion :

MqttConnectionService.cs

using System.Threading.Channels;
using Microsoft.Extensions.Options;
using MQTTnet;
using MQTTnet.Formatter;
using MQTTnet.Protocol;

public sealed class MqttConnectionService(
    IMqttClient client,
    IOptions<MqttSettings> settings,
    Channel<InboundMessage> inbox,
    ILogger<MqttConnectionService> log) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var s = settings.Value;
        var builder = new MqttClientOptionsBuilder()
            .WithTcpServer(s.Host, s.Port)
            .WithClientId(s.ClientId)
            .WithCredentials(s.Username, s.Password)
            .WithProtocolVersion(MqttProtocolVersion.V500)
            .WithKeepAlivePeriod(TimeSpan.FromSeconds(30));
        if (s.UseTls) builder.WithTlsOptions(o => o.UseTls());
        var options = builder.Build();

        client.ApplicationMessageReceivedAsync += async e =>
        {
            var m = e.ApplicationMessage;
            // On copie le nécessaire ; on ne garde pas d'objets MQTTnet au-delà du handler.
            await inbox.Writer.WriteAsync(
                new InboundMessage(m.Topic, m.ConvertPayloadToString(), DateTimeOffset.UtcNow), stoppingToken);
        };

        client.ConnectedAsync += async _ =>
        {
            var sub = new MqttClientSubscribeOptionsBuilder();
            foreach (var topic in s.Topics) sub.WithTopicFilter(topic, MqttQualityOfServiceLevel.AtLeastOnce);
            var result = await client.SubscribeAsync(sub.Build(), stoppingToken);
            foreach (var item in result.Items)
                log.LogInformation("SUBACK {Topic}: {Code}", item.TopicFilter.Topic, item.ResultCode);
        };

        client.DisconnectedAsync += e =>
        {
            if (e.ClientWasConnected) log.LogWarning(e.Exception, "MQTT disconnected: {Reason}", e.Reason);
            return Task.CompletedTask;
        };

        var delay = TimeSpan.FromSeconds(1);
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                if (!await client.TryPingAsync(stoppingToken))
                {
                    await client.ConnectAsync(options, stoppingToken);
                    log.LogInformation("MQTT connected to {Host}:{Port}", s.Host, s.Port);
                    delay = TimeSpan.FromSeconds(1);
                }
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { break; }
            catch (Exception ex)
            {
                log.LogWarning("MQTT connect failed: {Message}", ex.Message);
                delay = TimeSpan.FromSeconds(Math.Min(delay.TotalSeconds * 2, 60));
            }

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

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        await base.StopAsync(cancellationToken);
        if (client.IsConnected)
            await client.DisconnectAsync(new MqttClientDisconnectOptionsBuilder()
                .WithReason(MqttClientDisconnectOptionsReason.NormalDisconnection).Build(), cancellationToken);
        inbox.Writer.TryComplete();
    }
}

Comme le canal est borné avec FullMode = Wait, un consommateur lent fait attendre WriteAsync, ce qui retarde le PUBACK des messages QoS 1. Le broker cesse alors d'envoyer une fois le receive maximum du client atteint, au lieu que votre processus accumule des messages jusqu'à manquer de mémoire. Dimensionnez la capacité selon le nombre de secondes de trafic que vous acceptez de retenir.

StopAsync envoie un vrai DISCONNECT quand l'hôte s'arrête (Ctrl+C, arrêt du conteneur, déploiement). Sans cela, le broker ne remarque l'absence du client qu'à l'expiration du keep-alive et publie son Last Will, s'il y en a un.

Le consommateur et les handlers scoped

Si vous injectez un service scoped comme un DbContext directement dans un hosted service, l'app échoue au démarrage avec "Cannot consume scoped service 'AppDbContext' from singleton 'Microsoft.Extensions.Hosting.IHostedService'." La solution : un scope par message.

TelemetryConsumer.cs

using System.Threading.Channels;

public interface ITelemetryHandler
{
    Task HandleAsync(InboundMessage message, CancellationToken ct);
}

public sealed class TelemetryConsumer(
    Channel<InboundMessage> inbox,
    IServiceScopeFactory scopes,
    ILogger<TelemetryConsumer> log) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (var msg in inbox.Reader.ReadAllAsync(stoppingToken))
        {
            try
            {
                await using var scope = scopes.CreateAsyncScope();
                var handler = scope.ServiceProvider.GetRequiredService<ITelemetryHandler>();
                await handler.HandleAsync(msg, stoppingToken);
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                log.LogError(ex, "Failed to handle message on {Topic}", msg.Topic);
            }
        }
    }
}

public sealed class TelemetryHandler(ILogger<TelemetryHandler> log) : ITelemetryHandler
{
    public Task HandleAsync(InboundMessage message, CancellationToken ct)
    {
        log.LogInformation("{Topic}: {Payload}", message.Topic, message.Payload);
        return Task.CompletedTask; // enregistrer en base, appeler un autre service, ...
    }
}

Le try/catch autour de chaque message est indispensable : une exception non gérée dans ExecuteAsync arrête le hosted service — et, depuis .NET 6, l'hôte tout entier par défaut. Un payload malformé doit produire une ligne de log, pas une panne. Pour un traitement parallèle, lancez plusieurs boucles de lecture sur le même canal et passez SingleReader = false ; l'ordre par topic n'est alors plus garanti.

Health checks

MqttHealthCheck.cs

using Microsoft.Extensions.Diagnostics.HealthChecks;
using MQTTnet;

public sealed class MqttHealthCheck(IMqttClient client) : IHealthCheck
{
    public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken ct = default)
        => Task.FromResult(client.IsConnected
            ? HealthCheckResult.Healthy("Connected to MQTT broker")
            : new HealthCheckResult(context.Registration.FailureStatus, "Not connected to MQTT broker"));
}

Utiliser context.Registration.FailureStatus laisse l'enregistrement décider de la gravité d'une connexion perdue ; ici, il renvoie Degraded. Le check ne fait partie que de l'endpoint de readiness, tandis que /healthz/live n'exécute aucun check. Cette séparation est voulue : en cas de panne du broker, un orchestrateur qui redémarre tous les pods sur une liveness probe en échec transforme un problème en tempête de reconnexions. La readiness retire l'instance du load balancing pendant que la boucle de reconnexion fait son travail.

Publier depuis un endpoint d'API

Le client singleton peut être injecté partout. MQTTnet sérialise l'écriture des paquets en interne : des requêtes concurrentes peuvent donc publier sur la même connexion.

Program.cs — après app.Run()

static void MapCommandEndpoint(WebApplication app) =>
    app.MapPost("/devices/{id}/commands", async (string id, JsonElement body, IMqttClient mqtt, CancellationToken ct) =>
    {
        if (!Regex.IsMatch(id, "^[A-Za-z0-9_-]{1,64}$")) return Results.BadRequest("invalid device id");
        if (!mqtt.IsConnected) return Results.Problem("MQTT broker unavailable", statusCode: 503);

        var message = new MqttApplicationMessageBuilder()
            .WithTopic($"testmqtt/aspnet/devices/{id}/commands")
            .WithPayload(body.GetRawText())
            .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
            .WithMessageExpiryInterval(60)
            .Build();

        var result = await mqtt.PublishAsync(message, ct);
        return result.ReasonCode is MqttClientPublishReasonCode.Success or MqttClientPublishReasonCode.NoMatchingSubscribers
            ? Results.Accepted()
            : Results.Problem($"Broker rejected publish: {result.ReasonCode}", statusCode: 502);
    });

Valider l'ID n'est pas cosmétique : une valeur contenant +, # ou / modifierait la structure du topic. Avec MQTT 5, une publication QoS 1 vers un topic sans abonné renvoie NoMatchingSubscribers, ce qui n'est pas une erreur. Le message expiry empêche qu'un appareil qui se reconnecte bien plus tard exécute la commande. Utilisez le topic matcher pour vérifier que vos topics de commande correspondent à ceux auxquels les appareils s'abonnent.

Plusieurs instances

Avec deux réplicas et un abonnement classique, le broker livre chaque message aux deux, et chacun est traité deux fois. Utilisez une shared subscription dans le réglage Topics — $share/telemetry-workers/testmqtt/aspnet/telemetry/# — et le broker répartit les messages dans le groupe. Chaque réplica a toujours besoin de son propre client ID. Les détails et particularités de chaque broker sont dans shared subscriptions MQTT.

Pièges courants

Attendre la connexion au démarrage. Si vous vous connectez dans un constructeur ou bloquez StartAsync jusqu'à la réponse du broker, l'app ne démarre pas tant que le broker est injoignable. Laissez ExecuteAsync se connecter en arrière-plan ; le readiness check couvre l'intervalle.
S'abonner uniquement après la première connexion. Avec clean start, les abonnements disparaissent à chaque reconnexion. Abonnez-vous dans ConnectedAsync.
Attacher les handlers d'événements plusieurs fois. Si les événements sont attachés dans du code exécuté à chaque reconnexion, chaque message est traité une fois par attachement. Attachez-les exactement une fois.
Enregistrer IMqttClient en scoped ou transient. Chaque requête ouvrirait sa propre connexion au broker, ce qui est lent et atteint vite les limites de connexions. Gardez-le en singleton.

Tester le service

Lancez dotnet run et cherchez dans le log MQTT connected et une ligne SUBACK ... GrantedQoS1.
curl -i http://localhost:5000/healthz/ready doit renvoyer 200 Healthy (utilisez le port affiché par votre app).
Dans le client MQTT en ligne, publiez {"celsius":21.5} sur testmqtt/aspnet/telemetry/dev1 ; le handler le journalise.
Abonnez-vous à testmqtt/aspnet/devices/# dans le navigateur, lancez curl -X POST http://localhost:5000/devices/dev1/commands -H 'Content-Type: application/json' -d '{"led":"on"}' et regardez la commande arriver.
Bloquez un instant le port du broker (ou coupez le Wi-Fi) : la readiness passe à Degraded, puis revient à Healthy après la reconnexion.

Envoyer de la télémétrie de test à votre service ASP.NET Core

Si la connexion échoue avec une exception que vous ne connaissez pas, erreurs MQTTnet relie les messages courants à leurs causes.

Avec un client singleton, un service de connexion, un canal borné et des handlers scoped, MQTT devient une entrée comme une autre de votre app ASP.NET Core : configurée par options, observable par health checks, arrêtée proprement avec l'hôte et prête à monter en charge avec les shared subscriptions.

Questions fréquentes

Pourquoi l'erreur 'Cannot consume scoped service from singleton' dans mon hosted service MQTT ?

Les hosted services sont des singletons : ils ne peuvent pas recevoir de services scoped comme un DbContext dans leur constructeur. Injectez plutôt IServiceScopeFactory et créez un scope pour chaque message (ou lot) traité.

Le health check MQTT doit-il faire échouer la liveness probe ?

Généralement non. Si le broker est en panne, redémarrer votre app ne règle rien et peut déclencher une tempête de reconnexions. Exposez l'état MQTT sur un endpoint de readiness ou de statut et réservez la liveness aux problèmes qu'un redémarrage résout vraiment.

Comment lancer plusieurs instances sans traiter chaque message deux fois ?

Donnez à chaque instance un client ID unique et abonnez-vous via une shared subscription comme $share/telemetry-workers/testmqtt/aspnet/telemetry/#. Le broker répartit alors les messages entre les membres du groupe au lieu d'envoyer une copie à chaque instance.

MQTT dans ASP.NET Core : consommateur en hosted service, DI et health checks