Eine ASP.NET-Core-MQTT-Integration ist das natürliche Rückgrat eines .NET-IoT-Backends: Geräte veröffentlichen an einen Broker, und Ihre Web-App abonniert, speichert und liefert die Daten aus. In diesem Leitfaden betreiben Sie einen MQTTnet-Subscriber als BackgroundService, schreiben jeden Messwert mit EF Core in eine Datenbank und senden ihn mit SignalR in Echtzeit an Browser. Die fertige App läuft zum Testen gegen broker.connectmq.com und wechselt allein über die Konfiguration zu einem privaten ConnectMQ-Broker.
Architektur eines ASP.NET-Core-MQTT-Service
Der Aufbau der Lösung ist einfach, und jedes Teil hat genau eine Aufgabe:
Devices --MQTT--> Broker --MQTT--> MqttIngestService (BackgroundService)
|-- EF Core --> SQLite / SQL Server
|-- IHubContext --> SignalR --> Browser dashboard
Warum ein BackgroundService und kein Controller? MQTT ist eine langlebige, Push-basierte Verbindung. Ein Hosted Service startet mit der App, hält während seiner gesamten Lebensdauer eine Verbindung offen und beendet sich beim Herunterfahren sauber – genau der Lebenszyklus, den Sie brauchen. Controller sind an Requests gebunden und würden ständig Verbindungen öffnen und schließen.
Gerade im Mittelstand ist .NET häufig die Basis für MES-nahe Anwendungen, Leitstände und Energiemonitoring. Dort abonniert derselbe Service typischerweise einen Ausschnitt eines Unified Namespace (UNS), dessen Topics der ISA-95-Hierarchie folgen, etwa acme/werk-ulm/+/+/+/telemetry für alle Zellen eines Werks.
Zwei Designregeln sorgen für Robustheit:
- Der MQTT-Client ist ein Singleton im Besitz des Hosted Service. Erstellen Sie niemals einen Client pro Request.
- Datenbankkontexte sind scoped. Der Hosted Service ist ein Singleton und muss daher mit
IServiceScopeFactoryeinen Scope pro Nachrichten-Batch erzeugen.
Projekt-Setup
dotnet new web -n DeviceHub
cd DeviceHub
dotnet add package MQTTnet
dotnet add package Microsoft.EntityFrameworkCore.Sqlite
SignalR gehört zum Shared Framework von ASP.NET Core, ein zusätzliches Serverpaket ist also nicht nötig.
Fügen Sie die MQTT-Einstellungen in appsettings.json ein. Für den öffentlichen MQTT-Broker verwenden Sie eine beliebige eindeutige Client ID und ein zufälliges Topic-Präfix; er ist für alle lesbar und nur für Tests gedacht, senden Sie also nichts Vertrauliches und keine Produktionsdaten.
{
"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"
}
}
Das Datenmodell und der 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; } = "";
}
Der SignalR-Hub
Der Hub selbst kann leer bleiben. Der Background Service sendet über IHubContext<TelemetryHub>, und Browser hören nur zu.
using Microsoft.AspNetCore.SignalR;
public class TelemetryHub : Hub { }
Der MQTT Background Service
Das ist das Herzstück der ASP.NET-Core-MQTT-Integration. Er verbindet sich, abonniert, verbindet sich mit Backoff neu und verarbeitet Nachrichten, ohne die Empfangsschleife von MQTTnet zu blockieren.
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();
}
}
}
Einige Entscheidungen verdienen eine Erklärung:
- Der Message-Handler parst nur und reiht ein. Datenbank- und SignalR-Aufrufe erfolgen in einer separaten Schleife. Würde der Handler auf eine langsame Datenbank warten, hörte MQTTnet auf, den Socket zu lesen, Keep-alives könnten ablaufen und der Broker würde die Verbindung trennen.
- Ein begrenzter Channel mit
DropOldestdeckelt den Speicherverbrauch, falls die Datenbank ausfällt. Bei Telemetrie ist es meist besser, die ältesten Messwerte zu verlieren, als den Prozess abstürzen zu lassen. Wenn Sie keine Daten verlieren dürfen, wechseln Sie zuWaitund akzeptieren Backpressure. - Batching macht aus Hunderten Inserts einen einzigen
SaveChangesAsync-Aufruf. - Persistente Session (
CleanStart(false)plus 30 Minuten Session Expiry) bedeutet: QoS-1-Nachrichten, die während eines Neustarts Ihrer App veröffentlicht werden, kommen nach dem Reconnect an, und die Subscription bleibt bestehen. Das funktioniert nur mit stabiler Client ID – deshalb steht sie in der Konfiguration. - Fehlerhafte Payloads werden protokolliert und verworfen, nie erneut geworfen. Ein einzelnes defektes Gerät darf nicht die gesamte Datenaufnahme lahmlegen.
Verdrahtung in 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 reicht für eine Demo; in einem echten Projekt verwenden Sie EF-Core-Migrationen.
Ein minimales Live-Dashboard
Legen Sie wwwroot/index.html mit einer Liste und dem SignalR-JavaScript-Client aus einem CDN an. Das wesentliche Skript lautet:
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();
Beachten Sie die camelCase-Eigenschaftsnamen: Das JSON-Protokoll von SignalR serialisiert C#-Properties standardmäßig in camelCase.
Die Pipeline End-to-End testen
Starten Sie die App mit dotnet run, öffnen Sie die Seite und veröffentlichen Sie dann aus einem Terminal einen simulierten Messwert:
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}'
Der Messwert sollte binnen Sekundenbruchteilen im Browser erscheinen und unter /api/readings/boiler-1 abrufbar sein. Weitere Varianten für die Kommandozeile finden Sie in den Beispielen für mosquitto_pub und mosquitto_sub.
Befehle an Geräte zurücksenden
Die Datenaufnahme ist nur die Hälfte der meisten IoT-Backends; die andere Hälfte ist die Steuerung. Ein häufiger Fehler ist, in einem HTTP-Endpunkt für jeden Befehl eine neue MQTT-Verbindung zu öffnen. Das ist langsam, verschwendet Broker-Verbindungen (die auf das Kontingent Ihres Plans angerechnet werden) und scheitert unter Last. Halten Sie stattdessen einen zweiten langlebigen Client zum Veröffentlichen vor oder stellen Sie den Ingest-Client über einen kleinen Singleton-Wrapper bereit:
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);
}
Das Message-Expiry-Intervall ist bei Befehlen wichtig: Ist ein Gerät zehn Minuten offline, wollen Sie fast nie, dass es einen „Ventil öffnen“-Befehl in dem Moment ausführt, in dem es sich wieder verbindet. Mit 30 Sekunden Expiry verwirft der Broker den veralteten Befehl, statt ihn zuzustellen. Prüfen Sie den Reason Code im zurückgegebenen MqttClientPublishResult und melden Sie Fehler an den Aufrufer, statt für einen nie akzeptierten Befehl 200 OK zurückzugeben.
Ob Sie einen Client für Datenaufnahme und Befehle teilen oder zwei verwenden, ist eine Abwägung. Zwei Clients liefern übersichtlichere Logs und unabhängiges Reconnect-Verhalten; ein Client spart eine Verbindung. In einem kleinen Plan genügt meist einer.
Observability
Ein Background Service scheitert lautlos, wenn Sie ihn nicht laut machen. Protokollieren Sie mindestens Änderungen des Verbindungsstatus samt Grund, zählen Sie empfangene und verworfene Nachrichten pro Minute und erfassen Sie die Warteschlangentiefe des Channels. Der integrierte ILogger von ASP.NET Core und ein Health Check, der IsConnected meldet, decken die meisten Anforderungen ab; nutzen Sie bereits OpenTelemetry, stellen Sie diese Zähler als Metriken bereit, damit Sie alarmieren können, wenn die Datenaufnahme stoppt, während Geräte noch veröffentlichen.
Checkliste für den Produktivbetrieb
| Bereich | Was zu tun ist |
|---|---|
| Zugangsdaten | Benutzername und Passwort in User Secrets, Umgebungsvariablen oder einem Vault speichern, niemals in einer in Git eingecheckten appsettings.json |
| Client ID | Eine stabile ID pro laufender Instanz; zwei Instanzen mit derselben ID werfen sich gegenseitig hinaus |
| Horizontale Skalierung | Mehrere Instanzen mit einer normalen Subscription erhalten jeweils jede Nachricht; MQTT-5-Shared-Subscriptions ($share/devicehub/...) verteilen die Last |
| SignalR-Skalierung | Hinter mehreren Servern eine Backplane wie Redis oder Azure SignalR Service ergänzen |
| Health | IsConnected über einen ASP.NET-Core-Health-Check bereitstellen, damit Ihr Orchestrator Broker-Probleme erkennt |
| Payload-Validierung | Wertebereiche und Typen vor dem Speichern prüfen; fehlerhafte Daten ablehnen oder in Quarantäne verschieben |
Shared Subscriptions und QoS 2 sind ab dem kostenpflichtigen Serverless-Plan verfügbar; berücksichtigen Sie das, bevor Sie mehr als eine Ingest-Instanz betreiben. Der Leitfaden zu den MQTT-5-Funktionen erklärt Shared Subscriptions und Session Expiry ausführlicher.
Wechsel zu einem privaten Broker
Produktive Geräte sollten mit einem privaten, authentifizierten Broker kommunizieren. ConnectMQ bietet einen verwalteten MQTT-Broker mit TLS auf 8883, MQTT 3.1.1 und 5.0 sowie planabhängigen Kontingenten, sodass Ihre ASP.NET-Core-App nur die Konfiguration ändern muss:
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"
Da ConnectMQ Client IDs an die ausgestellten Zugangsdaten bindet, verwenden Sie für das Backend die im Portal generierte Client ID und geben jedem Gerät eigene Zugangsdaten. Wenn Sie die Client-Optionen genauer verstehen möchten, behandelt das MQTTnet-Tutorial TLS, MQTT-5-Properties und Reconnect-Details, und der Leitfaden zu MQTT-Client-Bibliotheken vergleicht .NET mit anderen Ökosystemen. Prüfen Sie noch grundsätzlich, wo der Broker laufen soll, lesen Sie über den Betrieb eines Cloud-MQTT-Brokers.
Starten Sie mit ConnectMQ
Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker, ganz ohne Kreditkarte, legen Sie die generierten Zugangsdaten in den User Secrets ab, und Ihr ASP.NET-Core-MQTT-Service ist vom ersten Tag an produktionsreif aufgebaut. Mit dem Live-Topic-Explorer im Portal beobachten Sie den Geräteverkehr, während Sie das Backend entwickeln.
Häufig gestellte Fragen
Sollte ich für MQTT IHostedService oder BackgroundService verwenden?
BackgroundService ist die einfachere Wahl, weil Sie eine einzige ExecuteAsync-Methode mit Stopping Token erhalten. Die Klasse implementiert IHostedService für Sie, daher werden beide gleich mit AddHostedService registriert.
Warum den DbContext nicht direkt in den Background Service injizieren?
Hosted Services sind Singletons, DbContext ist scoped. Eine direkte Injektion scheitert entweder an der Validierung oder hält einen Kontext dauerhaft am Leben; erzeugen Sie daher mit IServiceScopeFactory einen Scope pro Batch.
Wie betreibe ich mehrere Instanzen, ohne jede Nachricht doppelt zu verarbeiten?
Verwenden Sie eine MQTT-5-Shared-Subscription wie $share/devicehub/connectmq/+/telemetry. Der Broker stellt dann jede Nachricht nur einem Mitglied der Gruppe zu.
Kann ich aus einem Controller Befehle an Geräte veröffentlichen?
Ja. Registrieren Sie den verbundenen IMqttClient (oder einen kleinen Wrapper) als Singleton und rufen Sie PublishAsync aus Ihrem Endpunkt auf; erstellen Sie keine neue Verbindung pro HTTP-Request.
