MQTT en ASP.NET Core: BackgroundService, DI y health checks

Para consumir MQTT en ASP.NET Core, registra un único IMqttClient de MQTTnet en la inyección de dependencias, conéctalo y reconéctalo desde un BackgroundService, envía los mensajes entrantes a un Channel<T> y procésalos en un segundo hosted service que crea un scope de DI por mensaje. Un health check informa del estado de la conexión, y tus endpoints de API publican a través del mismo cliente.

Este tutorial construye ese servicio paso a paso con MQTTnet 5 sobre .NET 8 o posterior. Si no has usado MQTTnet antes, lee primero MQTT en C# con MQTTnet; este artículo se centra en la parte de hosting.

La arquitectura

Componente	Ciclo de vida	Función
`IMqttClient`	Singleton	Una conexión por instancia de la app, compartida por todo
`MqttConnectionService`	Hosted service	Conecta, reconecta, se suscribe y desconecta al apagar
`Channel<InboundMessage>`	Singleton	Buffer acotado entre la red y tu lógica de negocio
`TelemetryConsumer`	Hosted service	Lee el canal, crea un scope y llama a tu handler
`ITelemetryHandler`	Scoped	Tu código: puede usar `DbContext` y otros servicios scoped
`MqttHealthCheck`	Transient	Indica si el cliente está conectado

Separar la recepción del procesamiento importa. El event handler de MQTTnet se ejecuta en el dispatcher del cliente; si espera a una base de datos, todos los mensajes que vienen detrás esperan también, y la gestión del keep-alive puede resentirse. El canal desacopla ambas partes y te da backpressure de forma natural.

Proyecto y configuración

Crear el proyecto

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/#" ]
  }
}

Las credenciales viven en user secrets en local y en variables de entorno (Mqtt__Username, Mqtt__Password) o en un gestor de secretos en producción, nunca en appsettings.json. Vincula la sección a una clase de configuración:

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

El client ID por defecto usa el nombre de la máquina, que en Kubernetes es el nombre del pod y, por tanto, único por réplica. Dos instancias con el mismo client ID se expulsarían mutuamente del broker una y otra vez.

Registrar los servicios

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); // función local, ver más abajo
app.Run();

Con MQTTnet 4, sustituye MqttClientFactory por MqttFactory y añade using MQTTnet.Client;; el resto del código es igual.

El servicio de conexión

Este hosted service es dueño de la conexión. Reutiliza el bucle de reconexión del tutorial de MQTTnet y se vuelve a suscribir en ConnectedAsync, de modo que las suscripciones sobreviven a cada reconexión:

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;
            // Copiamos lo necesario; no guardamos objetos de MQTTnet fuera del 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();
    }
}

Como el canal está acotado con FullMode = Wait, un consumidor lento hace que WriteAsync espere, lo que retrasa el PUBACK de los mensajes QoS 1. El broker deja entonces de enviar cuando se alcanza el receive maximum del cliente, en lugar de que tu proceso acumule mensajes hasta quedarse sin memoria. Ajusta la capacidad a cuántos segundos de tráfico estás dispuesto a retener.

StopAsync envía un DISCONNECT correcto cuando el host se apaga (Ctrl+C, parada del contenedor, un despliegue). Sin él, el broker solo detecta la ausencia del cliente cuando expira el keep-alive y publica su Last Will, si hay uno configurado.

El consumidor y los handlers scoped

Si inyectas un servicio scoped como un DbContext directamente en un hosted service, la app falla al arrancar con "Cannot consume scoped service 'AppDbContext' from singleton 'Microsoft.Extensions.Hosting.IHostedService'." La solución es un scope por mensaje:

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; // guardar en base de datos, llamar a otro servicio, ...
    }
}

El try/catch alrededor de cada mensaje es esencial: una excepción no controlada en ExecuteAsync detiene el hosted service y, desde .NET 6, detiene por defecto todo el host. Un payload mal formado debe producir una entrada de log, no una caída. Si necesitas procesamiento en paralelo, lanza varios bucles lectores sobre el mismo canal y pon SingleReader = false; ten en cuenta que entonces ya no se garantiza el orden por topic.

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

Usar context.Registration.FailureStatus deja que el registro decida lo grave que es perder la conexión; aquí informa Degraded. El check solo forma parte del endpoint de readiness, mientras que /healthz/live no ejecuta ningún check. Esa separación es deliberada: si el broker sufre una caída, un orquestador que reinicia todos los pods por una liveness probe fallida convierte un problema en una tormenta de reconexiones. La readiness saca la instancia del balanceo mientras el bucle de reconexión hace su trabajo.

Publicar desde un endpoint de API

El cliente singleton se puede inyectar en cualquier parte. MQTTnet serializa internamente la escritura de paquetes, así que varias peticiones concurrentes pueden publicar por la misma conexión:

Program.cs — después de 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);
    });

Validar el ID no es cosmético: un valor con +, # o / cambiaría la estructura del topic. Con MQTT 5, una publicación QoS 1 a un topic sin suscriptores devuelve NoMatchingSubscribers, que no es un error. El message expiry evita que un dispositivo que reconecta mucho más tarde ejecute el comando. Usa el topic matcher para comprobar que tus topics de comandos coinciden con los que suscriben los dispositivos.

Ejecutar varias instancias

Con dos réplicas y una suscripción normal, el broker entrega cada mensaje a ambas y cada uno se procesa dos veces. Usa una shared subscription en la configuración Topics ($share/telemetry-workers/testmqtt/aspnet/telemetry/#) y el broker repartirá los mensajes entre el grupo. Cada réplica sigue necesitando su propio client ID. Los detalles y las particularidades de cada broker están en shared subscriptions de MQTT.

Errores comunes

Esperar a la conexión durante el arranque. Si conectas en un constructor o bloqueas StartAsync hasta que el broker responda, la app no arranca mientras el broker no sea accesible. Deja que ExecuteAsync conecte en segundo plano; el readiness check cubre ese intervalo.
Suscribirse solo tras la primera conexión. Con clean start las suscripciones desaparecen en cada reconexión. Suscríbete en ConnectedAsync.
Registrar los event handlers más de una vez. Si los eventos se registran en código que se ejecuta en cada reconexión, cada mensaje se procesa una vez por registro. Regístralos exactamente una vez.
Registrar IMqttClient como scoped o transient. Cada petición abriría su propia conexión al broker, lo que es lento y choca rápido con los límites de conexiones. Mantenlo como singleton.

Probar el servicio

Ejecuta dotnet run y busca en el log MQTT connected y una línea SUBACK ... GrantedQoS1.
curl -i http://localhost:5000/healthz/ready debería devolver 200 Healthy (usa el puerto que imprime tu app).
En el cliente MQTT online, publica {"celsius":21.5} en testmqtt/aspnet/telemetry/dev1; el handler lo registra en el log.
Suscríbete a testmqtt/aspnet/devices/# en el navegador, ejecuta curl -X POST http://localhost:5000/devices/dev1/commands -H 'Content-Type: application/json' -d '{"led":"on"}' y observa cómo llega el comando.
Bloquea el puerto del broker (o desconecta el Wi-Fi) un momento: la readiness pasa a Degraded y vuelve a Healthy tras la reconexión.

Envía telemetría de prueba a tu servicio ASP.NET Core

Si la conexión falla con una excepción que no reconoces, errores de MQTTnet relaciona los mensajes habituales con sus causas.

Con un cliente singleton, un servicio de conexión, un canal acotado y handlers scoped, MQTT se convierte en una entrada más de tu app ASP.NET Core: configurada mediante options, observable con health checks, apagada limpiamente con el host y lista para escalar con shared subscriptions.

Preguntas frecuentes

¿Por qué recibo 'Cannot consume scoped service from singleton' en mi hosted service MQTT?

Los hosted services son singletons, así que no pueden recibir servicios scoped como un DbContext en su constructor. Inyecta IServiceScopeFactory y crea un scope por cada mensaje (o lote) que proceses.

¿El health check de MQTT debe hacer fallar la liveness probe?

Normalmente no. Si el broker está caído, reiniciar tu app no lo arregla y puede provocar una tormenta de reconexiones. Informa de la conexión MQTT en un endpoint de readiness o de estado y deja la liveness para problemas que un reinicio sí resuelve.

¿Cómo ejecuto varias instancias sin procesar cada mensaje dos veces?

Da a cada instancia un client ID único y suscríbete con una shared subscription como $share/telemetry-workers/testmqtt/aspnet/telemetry/#. El broker reparte entonces los mensajes entre el grupo en lugar de enviar una copia a cada instancia.

MQTT en ASP.NET Core: consumidor con hosted service, DI y health checks