MQTT no ASP.NET Core: BackgroundService, DI e health checks

Para consumir MQTT no ASP.NET Core, registre um único IMqttClient do MQTTnet na injeção de dependência, conecte e reconecte a partir de um BackgroundService, envie as mensagens recebidas para um Channel<T> e processe-as num segundo hosted service que cria um scope de DI por mensagem. Um health check informa o estado da conexão, e os endpoints da sua API publicam pelo mesmo cliente.

Este tutorial monta esse serviço passo a passo com MQTTnet 5 no .NET 8 ou superior. Se você nunca usou o MQTTnet, leia antes MQTT em C# com MQTTnet; este artigo foca na parte de hosting.

A arquitetura

Componente	Ciclo de vida	Função
`IMqttClient`	Singleton	Uma conexão por instância do app, compartilhada por tudo
`MqttConnectionService`	Hosted service	Conecta, reconecta, assina e desconecta no shutdown
`Channel<InboundMessage>`	Singleton	Buffer limitado entre a rede e a lógica de negócio
`TelemetryConsumer`	Hosted service	Lê o canal, cria um scope e chama o seu handler
`ITelemetryHandler`	Scoped	Seu código — pode usar `DbContext` e outros serviços scoped
`MqttHealthCheck`	Transient	Informa se o cliente está conectado

Separar recebimento de processamento faz diferença. O event handler do MQTTnet roda no dispatcher do cliente; se ele espera um banco de dados, todas as mensagens atrás dele esperam também, e o tratamento do keep-alive pode sofrer. O canal desacopla as duas partes e oferece backpressure natural.

Projeto e configuração

Criar o projeto

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

As credenciais ficam em user secrets localmente e em variáveis de ambiente (Mqtt__Username, Mqtt__Password) ou num cofre de segredos em produção — nunca no appsettings.json. Vincule a seção a uma classe de configuração:

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

O client ID padrão usa o nome da máquina, que no Kubernetes é o nome do pod e, portanto, único por réplica. Duas instâncias com o mesmo client ID ficariam derrubando uma à outra do broker.

Registrar os serviços

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); // função local, veja abaixo
app.Run();

No MQTTnet 4, troque MqttClientFactory por MqttFactory e adicione using MQTTnet.Client; — o restante do código é igual.

O serviço de conexão

Este hosted service é dono da conexão. Ele reaproveita o loop de reconexão do tutorial de MQTTnet e refaz as assinaturas em ConnectedAsync, para que elas sobrevivam a cada reconexão:

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 o necessário; não guardamos objetos do MQTTnet além do 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 o canal é limitado com FullMode = Wait, um consumidor lento faz o WriteAsync esperar, o que atrasa o PUBACK das mensagens QoS 1. O broker então para de enviar quando o receive maximum do cliente é atingido, em vez de o seu processo acumular mensagens até ficar sem memória. Ajuste a capacidade a quantos segundos de tráfego você aceita reter.

O StopAsync envia um DISCONNECT correto quando o host desliga (Ctrl+C, parada do contêiner, um deploy). Sem ele, o broker só percebe a ausência do cliente quando o keep-alive expira e publica o Last Will, se houver um configurado.

O consumidor e os handlers scoped

Se você injetar um serviço scoped como um DbContext diretamente num hosted service, o app falha na inicialização com "Cannot consume scoped service 'AppDbContext' from singleton 'Microsoft.Extensions.Hosting.IHostedService'." A solução é um scope por mensagem:

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; // salvar no banco de dados, chamar outro serviço, ...
    }
}

O try/catch em volta de cada mensagem é essencial: uma exceção não tratada em ExecuteAsync para o hosted service — e, desde o .NET 6, para o host inteiro por padrão. Um payload malformado deve gerar uma entrada de log, não uma indisponibilidade. Se precisar de processamento paralelo, inicie vários loops de leitura no mesmo canal e defina SingleReader = false; lembre que a ordem por tópico deixa de ser garantida.

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 deixa o registro decidir a gravidade de uma conexão perdida; aqui ele reporta Degraded. O check faz parte apenas do endpoint de readiness, enquanto /healthz/live não executa nenhum check. Essa divisão é proposital: se o broker cair, um orquestrador que reinicia todos os pods por uma liveness probe com falha transforma um problema numa tempestade de reconexões. A readiness tira a instância do balanceamento enquanto o loop de reconexão faz o trabalho dele.

Publicar a partir de um endpoint da API

O cliente singleton pode ser injetado em qualquer lugar. O MQTTnet serializa internamente a escrita dos pacotes, então requisições concorrentes podem publicar pela mesma conexão:

Program.cs — depois 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 o ID não é detalhe: um valor com +, # ou / mudaria a estrutura do tópico. Com MQTT 5, um publish QoS 1 para um tópico sem assinantes retorna NoMatchingSubscribers, o que não é erro. O message expiry impede que um dispositivo que reconecta muito depois execute o comando. Use o topic matcher para conferir se seus tópicos de comando casam com o que os dispositivos assinam.

Rodando várias instâncias

Com duas réplicas e uma assinatura normal, o broker entrega cada mensagem para as duas, e cada uma é processada duas vezes. Use uma shared subscription na configuração Topics — $share/telemetry-workers/testmqtt/aspnet/telemetry/# — e o broker distribui as mensagens entre o grupo. Cada réplica continua precisando do próprio client ID. Os detalhes e as particularidades de cada broker estão em shared subscriptions no MQTT.

Armadilhas comuns

Esperar a conexão durante a inicialização. Se você conecta num construtor ou bloqueia o StartAsync até o broker responder, o app não sobe enquanto o broker estiver inacessível. Deixe o ExecuteAsync conectar em segundo plano; o readiness check cobre esse intervalo.
Assinar só depois da primeira conexão. Com clean start as assinaturas somem a cada reconexão. Assine em ConnectedAsync.
Registrar event handlers mais de uma vez. Se os eventos são registrados num código que roda a cada reconexão, cada mensagem é processada uma vez por registro. Registre-os exatamente uma vez.
Registrar IMqttClient como scoped ou transient. Cada requisição abriria a própria conexão com o broker, o que é lento e logo esbarra nos limites de conexão. Mantenha como singleton.

Testar o serviço

Rode dotnet run e procure no log MQTT connected e uma linha SUBACK ... GrantedQoS1.
curl -i http://localhost:5000/healthz/ready deve retornar 200 Healthy (use a porta que o app imprime).
No cliente MQTT online, publique {"celsius":21.5} em testmqtt/aspnet/telemetry/dev1; o handler registra no log.
Assine testmqtt/aspnet/devices/# no navegador, rode curl -X POST http://localhost:5000/devices/dev1/commands -H 'Content-Type: application/json' -d '{"led":"on"}' e veja o comando chegar.
Bloqueie a porta do broker (ou desligue o Wi-Fi) por um instante: a readiness passa para Degraded e volta para Healthy após a reconexão.

Envie telemetria de teste para o seu serviço ASP.NET Core

Se a conexão falhar com uma exceção que você não reconhece, erros do MQTTnet relaciona as mensagens comuns às suas causas.

Com um cliente singleton, um serviço de conexão, um canal limitado e handlers scoped, o MQTT vira só mais uma entrada do seu app ASP.NET Core: configurado por options, observável por health checks, desligado de forma limpa junto com o host e pronto para escalar com shared subscriptions.

Perguntas frequentes

Por que recebo 'Cannot consume scoped service from singleton' no meu hosted service MQTT?

Hosted services são singletons, então não podem receber serviços scoped como um DbContext no construtor. Injete IServiceScopeFactory e crie um scope para cada mensagem (ou lote) processada.

O health check do MQTT deve derrubar a liveness probe?

Normalmente não. Se o broker estiver fora do ar, reiniciar o app não resolve e pode causar uma tempestade de reconexões. Reporte a conexão MQTT num endpoint de readiness ou de status e deixe a liveness para problemas que um restart realmente resolve.

Como rodar várias instâncias sem processar cada mensagem duas vezes?

Dê a cada instância um client ID único e assine com uma shared subscription como $share/telemetry-workers/testmqtt/aspnet/telemetry/#. O broker passa a distribuir as mensagens entre o grupo em vez de mandar uma cópia para cada instância.

MQTT no ASP.NET Core: consumidor com hosted service, DI e health checks