ASP.NET Core MQTT : un service d'arrière-plan qui stocke et diffuse les données des appareils

Une intégration ASP.NET Core MQTT constitue l'épine dorsale naturelle d'un backend IoT en .NET : les appareils publient sur un broker, et votre application web s'abonne, stocke et sert les données. Dans ce guide, vous allez héberger un subscriber MQTTnet en tant que BackgroundService, écrire chaque relevé en base de données avec EF Core et le pousser vers les navigateurs en temps réel avec SignalR. L'application finale fonctionne sur le serveur MQTT public gratuit broker.connectmq.com pour les tests et bascule vers un broker ConnectMQ privé par simple configuration.

Architecture d'un service ASP.NET Core MQTT

La structure de la solution est simple, et chaque élément a un seul rôle :

Devices --MQTT--> Broker --MQTT--> MqttIngestService (BackgroundService)
                                     |-- EF Core --> SQLite / SQL Server
                                     |-- IHubContext --> SignalR --> Browser dashboard

Pourquoi un BackgroundService plutôt qu'un contrôleur ? MQTT repose sur une connexion longue durée en mode push. Un hosted service démarre avec l'application, maintient une connexion ouverte pendant toute sa durée de vie et s'arrête proprement à l'extinction : c'est exactement le cycle de vie recherché. Les contrôleurs ont une portée limitée à la requête et ouvriraient et fermeraient des connexions en permanence.

Deux règles de conception assurent la robustesse :

1. Le client MQTT est un singleton détenu par le hosted service. Ne créez jamais un client par requête.
2. Les contextes de base de données sont scoped. Le hosted service étant un singleton, il doit créer un scope par lot de messages avec IServiceScopeFactory.

Configuration du projet

dotnet new web -n DeviceHub
cd DeviceHub
dotnet add package MQTTnet
dotnet add package Microsoft.EntityFrameworkCore.Sqlite

SignalR fait partie du framework partagé ASP.NET Core : aucun package serveur supplémentaire n'est nécessaire.

Ajoutez les paramètres MQTT dans appsettings.json. Pour le serveur MQTT public gratuit, utilisez n'importe quel client ID unique et un préfixe de topic aléatoire ; ce serveur est lisible par tous et réservé aux tests, n'envoyez donc aucune donnée sensible ou de production.

{
  "Mqtt": {
    "Host": "broker.connectmq.com",
    "Port": 8883,
    "UseTls": true,
    "ClientId": "devicehub-8d21c4",
    "Username": "",
    "Password": "",
    "TopicFilter": "connectmq/demo/8d21c4/+/telemetry"
  },
  "ConnectionStrings": {
    "Default": "Data Source=devicehub.db"
  }
}

Le modèle de données et le DbContext

using Microsoft.EntityFrameworkCore;

public class Reading
{
    public long Id { get; set; }
    public string DeviceId { get; set; } = "";
    public double Temperature { get; set; }
    public DateTimeOffset ReceivedAt { get; set; }
}

public class DeviceDb(DbContextOptions<DeviceDb> options) : DbContext(options)
{
    public DbSet<Reading> Readings => Set<Reading>();
}

public class MqttOptions
{
    public string Host { get; set; } = "";
    public int Port { get; set; } = 8883;
    public bool UseTls { get; set; } = true;
    public string ClientId { get; set; } = "";
    public string? Username { get; set; }
    public string? Password { get; set; }
    public string TopicFilter { get; set; } = "";
}

Le hub SignalR

Le hub lui-même peut rester vide. Le service d'arrière-plan pousse les données via IHubContext<TelemetryHub>, et les navigateurs se contentent d'écouter.

using Microsoft.AspNetCore.SignalR;

public class TelemetryHub : Hub { }

Le service d'arrière-plan MQTT

C'est le cœur de l'intégration ASP.NET Core MQTT. Il se connecte, s'abonne, se reconnecte avec backoff et traite les messages sans bloquer la boucle de réception de MQTTnet.

using System.Text.Json;
using System.Threading.Channels;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Options;
using MQTTnet;
using MQTTnet.Formatter;
using MQTTnet.Protocol;

public class MqttIngestService(
    IOptions<MqttOptions> options,
    IServiceScopeFactory scopeFactory,
    IHubContext<TelemetryHub> hub,
    ILogger<MqttIngestService> logger) : BackgroundService
{
    private readonly MqttOptions _opt = options.Value;
    private readonly Channel<Reading> _queue =
        Channel.CreateBounded<Reading>(new BoundedChannelOptions(10_000)
        { FullMode = BoundedChannelFullMode.DropOldest });

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var client = new MqttClientFactory().CreateMqttClient();
        client.ApplicationMessageReceivedAsync += OnMessageAsync;

        var builder = new MqttClientOptionsBuilder()
            .WithTcpServer(_opt.Host, _opt.Port)
            .WithClientId(_opt.ClientId)
            .WithProtocolVersion(MqttProtocolVersion.V500)
            .WithCleanStart(false)
            .WithSessionExpiryInterval(1800);
        if (_opt.UseTls) builder.WithTlsOptions(o => o.UseTls());
        if (!string.IsNullOrEmpty(_opt.Username))
            builder.WithCredentials(_opt.Username, _opt.Password);
        var mqttOptions = builder.Build();

        var writer = Task.Run(() => WriteLoopAsync(stoppingToken), stoppingToken);
        var delay = TimeSpan.FromSeconds(1);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                if (!await client.TryPingAsync(stoppingToken))
                {
                    var result = await client.ConnectAsync(mqttOptions, stoppingToken);
                    logger.LogInformation("MQTT connected, session present: {Present}",
                        result.IsSessionPresent);
                    if (!result.IsSessionPresent)
                    {
                        await client.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
                            .WithTopicFilter(f => f.WithTopic(_opt.TopicFilter)
                                .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
                            .Build(), stoppingToken);
                    }
                    delay = TimeSpan.FromSeconds(1);
                }
            }
            catch (OperationCanceledException) { break; }
            catch (Exception ex)
            {
                logger.LogWarning("MQTT connect failed: {Message}", ex.Message);
                delay = TimeSpan.FromSeconds(Math.Min(delay.TotalSeconds * 2, 60));
            }

            try { await Task.Delay(delay, stoppingToken); }
            catch (OperationCanceledException) { break; }
        }

        if (client.IsConnected) await client.DisconnectAsync();
        client.Dispose();
        _queue.Writer.TryComplete();
        await writer;
    }

    private Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs e)
    {
        try
        {
            // Topic: connectmq/demo/<prefix>/<deviceId>/telemetry
            var parts = e.ApplicationMessage.Topic.Split('/');
            var deviceId = parts[^2];
            using var doc = JsonDocument.Parse(e.ApplicationMessage.ConvertPayloadToString());
            var reading = new Reading
            {
                DeviceId = deviceId,
                Temperature = doc.RootElement.GetProperty("temperature").GetDouble(),
                ReceivedAt = DateTimeOffset.UtcNow
            };
            _queue.Writer.TryWrite(reading);
        }
        catch (Exception ex)
        {
            logger.LogWarning("Dropping bad payload on {Topic}: {Message}",
                e.ApplicationMessage.Topic, ex.Message);
        }
        return Task.CompletedTask;
    }

    private async Task WriteLoopAsync(CancellationToken ct)
    {
        var batch = new List<Reading>(100);
        await foreach (var first in _queue.Reader.ReadAllAsync(CancellationToken.None))
        {
            batch.Add(first);
            while (batch.Count < 100 && _queue.Reader.TryRead(out var next))
                batch.Add(next);

            using (var scope = scopeFactory.CreateScope())
            {
                var db = scope.ServiceProvider.GetRequiredService<DeviceDb>();
                db.Readings.AddRange(batch);
                await db.SaveChangesAsync(CancellationToken.None);
            }

            foreach (var r in batch)
                await hub.Clients.All.SendAsync("reading", r, CancellationToken.None);

            batch.Clear();
        }
    }
}

Quelques choix méritent une explication :

• Le handler de messages se contente d'analyser et de mettre en file. Les appels à la base de données et à SignalR se font dans une boucle séparée. Si le handler attendait une base lente, MQTTnet cesserait de lire le socket, les keep-alives pourraient expirer et le broker couperait la connexion.
• Un canal borné avec DropOldest plafonne la mémoire si la base de données tombe. Pour de la télémétrie, perdre les relevés les plus anciens vaut généralement mieux que faire planter le processus. Si vous ne pouvez pas perdre de données, passez à Wait et acceptez la contre-pression.
• Le traitement par lots transforme des centaines d'insertions en un seul appel à SaveChangesAsync.
• La session persistante (CleanStart(false) plus une expiration de session de 30 minutes) garantit que les messages QoS 1 publiés pendant le redémarrage de votre application sont livrés à la reconnexion, et que l'abonnement survit. Cela ne fonctionne que si le client ID reste stable, d'où sa place dans la configuration.
• Les payloads malformés sont journalisés et abandonnés, jamais relancés. Un seul appareil défaillant ne doit pas faire tomber l'ingestion.

Câblage dans Program.cs

using Microsoft.EntityFrameworkCore;

var builder = WebApplication.CreateBuilder(args);

builder.Services.Configure<MqttOptions>(builder.Configuration.GetSection("Mqtt"));
builder.Services.AddDbContext<DeviceDb>(o =>
    o.UseSqlite(builder.Configuration.GetConnectionString("Default")));
builder.Services.AddSignalR();
builder.Services.AddHostedService<MqttIngestService>();

var app = builder.Build();

using (var scope = app.Services.CreateScope())
    scope.ServiceProvider.GetRequiredService<DeviceDb>().Database.EnsureCreated();

app.UseDefaultFiles();
app.UseStaticFiles();
app.MapHub<TelemetryHub>("/hubs/telemetry");

app.MapGet("/api/readings/{deviceId}", async (string deviceId, DeviceDb db) =>
    await db.Readings.Where(r => r.DeviceId == deviceId)
        .OrderByDescending(r => r.Id).Take(50).ToListAsync());

app.Run();

EnsureCreated convient pour une démo ; utilisez les migrations EF Core dans un vrai projet.

Un tableau de bord en direct minimal

Créez wwwroot/index.html avec une liste et le client JavaScript SignalR chargé depuis un CDN. Le script essentiel est :

const connection = new signalR.HubConnectionBuilder()
  .withUrl("/hubs/telemetry")
  .withAutomaticReconnect()
  .build();

connection.on("reading", r => {
  const li = document.createElement("li");
  li.textContent = `${r.receivedAt} ${r.deviceId}: ${r.temperature} °C`;
  document.getElementById("log").prepend(li);
});

connection.start();

Notez les noms de propriétés en camelCase : le protocole JSON de SignalR sérialise par défaut les propriétés C# en camelCase.

Tester le pipeline de bout en bout

Lancez l'application avec dotnet run, ouvrez la page, puis publiez un faux relevé depuis un terminal :

mosquitto_pub -h broker.connectmq.com -p 1883 -i devicehub-test-1 \
  -t connectmq/demo/8d21c4/boiler-1/telemetry -q 1 \
  -m '{"temperature": 71.4}'

Le relevé devrait apparaître dans le navigateur en une fraction de seconde et être consultable sur /api/readings/boiler-1. Pour d'autres variantes en ligne de commande, consultez les exemples mosquitto_pub et mosquitto_sub.

Renvoyer des commandes aux appareils

L'ingestion ne représente que la moitié de la plupart des backends IoT ; l'autre moitié, c'est le pilotage. Une erreur courante consiste à ouvrir une nouvelle connexion MQTT dans un endpoint HTTP pour chaque commande. C'est lent, cela gaspille des connexions au broker (qui comptent dans le quota de votre plan) et cela échoue sous charge. Conservez plutôt un second client longue durée dédié à la publication, ou exposez le client d'ingestion via un petit wrapper singleton :

public class MqttCommandPublisher(IMqttClient client)
{
    public Task<MqttClientPublishResult> SendAsync(string deviceId, object command,
        CancellationToken ct) =>
        client.PublishAsync(new MqttApplicationMessageBuilder()
            .WithTopic($"connectmq/demo/8d21c4/{deviceId}/cmd")
            .WithPayload(JsonSerializer.Serialize(command))
            .WithContentType("application/json")
            .WithMessageExpiryInterval(30)
            .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
            .Build(), ct);
}

L'intervalle d'expiration du message est important pour les commandes : si un appareil est hors ligne pendant dix minutes, vous ne voulez presque jamais qu'il exécute une commande « ouvrir la vanne » au moment où il se reconnecte. Avec une expiration de 30 secondes, le broker supprime la commande périmée au lieu de la livrer. Vérifiez le reason code dans le MqttClientPublishResult renvoyé et remontez les échecs à l'appelant, plutôt que de renvoyer 200 OK pour une commande qui n'a jamais été acceptée.

Partager un seul client entre ingestion et commandes ou en utiliser deux relève de l'appréciation. Deux clients offrent des journaux plus clairs et des reconnexions indépendantes ; un seul client économise une connexion. Sur un petit plan, un seul suffit généralement.

Observabilité

Un service d'arrière-plan échoue en silence, à moins que vous ne le rendiez bavard. Au minimum, journalisez les changements d'état de la connexion avec leur raison, comptez les messages reçus et abandonnés par minute, et enregistrez la profondeur de file du canal. L'ILogger intégré d'ASP.NET Core et un health check qui expose IsConnected couvrent la plupart des besoins ; si vous utilisez déjà OpenTelemetry, exposez ces compteurs sous forme de métriques afin de recevoir une alerte lorsque l'ingestion s'arrête alors que les appareils publient toujours.

Check-list de production

Les shared subscriptions et la QoS 2 sont disponibles à partir du plan Serverless payant : anticipez-le avant d'exécuter plus d'une instance d'ingestion. Le guide des fonctionnalités de MQTT 5 détaille les shared subscriptions et l'expiration de session.

Passer à un broker privé

Les appareils de production doivent communiquer avec un broker privé et authentifié. ConnectMQ fournit un broker MQTT managé avec TLS sur le port 8883, MQTT 3.1.1 et 5.0, et des quotas par plan : votre application ASP.NET Core n'a donc qu'à changer de configuration :

dotnet user-secrets set "Mqtt:Host" "YOUR_BROKER_HOST"
dotnet user-secrets set "Mqtt:ClientId" "YOUR_CLIENT_ID"
dotnet user-secrets set "Mqtt:Username" "YOUR_USERNAME"
dotnet user-secrets set "Mqtt:Password" "YOUR_PASSWORD"

Comme ConnectMQ lie les client ID aux identifiants émis, utilisez pour le backend le client ID généré dans le portail, et attribuez à chaque appareil ses propres identifiants. Pour approfondir les options du client, le tutoriel MQTTnet couvre TLS, les propriétés MQTT 5 et les détails de reconnexion, et le guide des bibliothèques client MQTT compare .NET aux autres écosystèmes. Si vous vous demandez encore où héberger le broker, lisez notre article sur l'exploitation d'un broker MQTT dans le cloud.

Démarrer avec ConnectMQ

Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, placez les identifiants générés dans les user secrets, et votre service ASP.NET Core MQTT est prêt pour la production dès le premier jour. Utilisez l'explorateur de topics en direct du portail pour observer le trafic des appareils pendant que vous développez le backend.

Questions fréquentes

Faut-il utiliser IHostedService ou BackgroundService pour MQTT ?

BackgroundService est le choix le plus simple, car il vous fournit une unique méthode ExecuteAsync avec un jeton d'arrêt. Il implémente IHostedService pour vous : les deux s'enregistrent donc de la même façon avec AddHostedService.

Pourquoi ne pas injecter directement le DbContext dans le service d'arrière-plan ?

Les hosted services sont des singletons et DbContext est scoped. L'injecter directement échoue à la validation ou maintient un même contexte en vie indéfiniment : créez donc un scope par lot avec IServiceScopeFactory.

Comment exécuter plusieurs instances sans traiter chaque message deux fois ?

Utilisez une shared subscription MQTT 5 comme $share/devicehub/connectmq/+/telemetry. Le broker livre alors chaque message à un seul membre du groupe.

Puis-je publier des commandes vers les appareils depuis un contrôleur ?

Oui. Enregistrez l'IMqttClient connecté (ou un petit wrapper) comme singleton et appelez PublishAsync depuis votre endpoint ; ne créez pas une nouvelle connexion par requête HTTP.