MQTT com ASP.NET Core: um Background Service que Armazena e Transmite Dados de Dispositivos

Uma integração ASP.NET Core MQTT é a espinha dorsal natural de um backend IoT em .NET: os dispositivos publicam em um broker, e a sua aplicação web assina, armazena e entrega os dados. Neste guia você vai hospedar um subscriber MQTTnet como um BackgroundService, gravar cada leitura em um banco de dados com EF Core e enviá-la aos navegadores em tempo real com SignalR. A aplicação final roda no servidor MQTT público broker.connectmq.com para testes e passa para um broker privado da ConnectMQ apenas com mudanças de configuração.

Arquitetura de um serviço ASP.NET Core MQTT

O formato da solução é simples, e cada peça tem uma única função:

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

Por que um BackgroundService e não um controller? O MQTT é uma conexão de longa duração baseada em push. Um hosted service inicia com a aplicação, mantém uma conexão aberta durante todo o seu ciclo de vida e para de forma ordenada no encerramento, que é exatamente o ciclo de vida desejado. Os controllers têm escopo de requisição e ficariam abrindo e fechando conexões o tempo todo.

Duas regras de design mantêm isso robusto:

1. O cliente MQTT é um singleton pertencente ao hosted service. Nunca crie um cliente por requisição.
2. Os contextos de banco de dados têm escopo. O hosted service é um singleton, então ele precisa criar um escopo por lote de mensagens com IServiceScopeFactory.

Configuração do projeto

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

O SignalR faz parte do framework compartilhado do ASP.NET Core, então nenhum pacote extra de servidor é necessário.

Adicione as configurações de MQTT ao appsettings.json. Para o servidor MQTT público, use qualquer client ID único e um prefixo de topic aleatório; ele pode ser lido por qualquer pessoa e serve apenas para testes, então não envie nada sensível.

{
  "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"
  }
}

O modelo de dados e o 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; } = "";
}

O hub do SignalR

O hub em si pode ficar vazio. O background service envia por meio de IHubContext<TelemetryHub>, e os navegadores apenas escutam.

using Microsoft.AspNetCore.SignalR;

public class TelemetryHub : Hub { }

O background service MQTT

Este é o coração da integração ASP.NET Core MQTT. Ele conecta, assina, reconecta com back-off e processa as mensagens sem bloquear o loop de recebimento do 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();
        }
    }
}

Algumas escolhas que vale explicar:

• O handler de mensagens apenas faz o parse e enfileira. As chamadas ao banco de dados e ao SignalR acontecem em um loop separado. Se o handler aguardasse um banco lento, o MQTTnet pararia de ler o socket, os keep-alives poderiam expirar e o broker derrubaria a conexão.
• Um channel limitado com DropOldest limita o uso de memória se o banco de dados cair. Para telemetria, perder as leituras mais antigas costuma ser melhor do que derrubar o processo. Se você não pode perder dados, mude para Wait e aceite o backpressure.
• O processamento em lotes transforma centenas de inserts em uma única chamada a SaveChangesAsync.
• A sessão persistente (CleanStart(false) mais uma expiração de sessão de 30 minutos) faz com que as mensagens QoS 1 publicadas enquanto a sua aplicação reinicia sejam entregues quando ela reconectar, e a assinatura sobrevive. Isso só funciona se o client ID permanecer estável, e é por isso que ele fica na configuração.
• Payloads malformados são registrados em log e descartados, nunca relançados. Um único dispositivo com problema não deve derrubar a ingestão.

Juntando tudo no 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();

O EnsureCreated serve para uma demonstração; use as migrations do EF Core em um projeto real.

Um dashboard ao vivo mínimo

Crie wwwroot/index.html com uma lista e o cliente JavaScript do SignalR carregado de uma CDN. O script essencial é:

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

Observe os nomes de propriedades em camelCase: o protocolo JSON do SignalR serializa as propriedades C# em camelCase por padrão.

Testando o pipeline de ponta a ponta

Execute a aplicação com dotnet run, abra a página e publique uma leitura falsa, como a de uma caldeira de frigorífico, a partir de um 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}'

A leitura deve aparecer no navegador em uma fração de segundo e estar disponível em /api/readings/boiler-1. Para mais variações de linha de comando, veja os exemplos de mosquitto_pub e mosquitto_sub.

Enviando comandos de volta aos dispositivos

A ingestão é só metade da maioria dos backends IoT; a outra metade é o controle. Um erro comum é abrir uma nova conexão MQTT dentro de um endpoint HTTP a cada comando. Isso é lento, desperdiça conexões do broker (que contam na cota do seu plano) e falha sob carga. Em vez disso, mantenha um segundo cliente de longa duração para publicar ou exponha o cliente de ingestão por meio de um pequeno 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);
}

O intervalo de expiração de mensagem é importante para comandos: se um dispositivo ficar offline por dez minutos, quase nunca você vai querer que ele execute um comando "abrir válvula" no instante em que reconectar. Com uma expiração de 30 segundos, o broker descarta o comando obsoleto em vez de entregá-lo. Verifique o reason code no MqttClientPublishResult retornado e repasse as falhas a quem chamou, em vez de retornar 200 OK para um comando que nunca foi aceito.

Compartilhar um cliente entre ingestão e comandos ou usar dois é uma questão de julgamento. Dois clientes oferecem logs mais claros e comportamento de reconexão independente; um único cliente economiza uma conexão. Em um plano pequeno, um costuma bastar.

Observabilidade

Um background service falha em silêncio, a menos que você o faça falar alto. No mínimo, registre as mudanças de estado da conexão com o motivo, conte as mensagens recebidas e descartadas por minuto e registre a profundidade da fila do channel. O ILogger embutido do ASP.NET Core e um health check que informe IsConnected cobrem a maioria das necessidades; se você já usa OpenTelemetry, exponha esses contadores como métricas para poder disparar alertas quando a ingestão parar enquanto os dispositivos ainda estão publicando.

Checklist de produção

Shared subscriptions e QoS 2 estão disponíveis a partir do plano Serverless pago, então planeje isso antes de rodar mais de uma instância de ingestão. O guia de recursos do MQTT 5 explica shared subscriptions e expiração de sessão em mais detalhes.

Migrando para um broker privado

Dispositivos em produção devem falar com um broker privado e autenticado. A ConnectMQ oferece um broker MQTT gerenciado com TLS na porta 8883, MQTT 3.1.1 e 5.0 e cotas por plano, então a sua aplicação ASP.NET Core só precisa mudar a configuração:

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"

Como a ConnectMQ vincula os client IDs às credenciais emitidas, use o client ID gerado no portal para o backend e dê a cada dispositivo suas próprias credenciais. Se quiser entender as opções do cliente com mais profundidade, o tutorial de MQTTnet aborda TLS, propriedades do MQTT 5 e detalhes de reconexão, e o guia de bibliotecas de cliente MQTT compara o .NET com outros ecossistemas. Se você ainda está avaliando onde hospedar o broker, leia sobre como rodar um broker MQTT na nuvem.

Comece a usar a ConnectMQ

Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, coloque as credenciais geradas nos user secrets e o seu serviço ASP.NET Core MQTT estará com cara de produção desde o primeiro dia. Use o explorador de topics em tempo real do portal para acompanhar o tráfego dos dispositivos enquanto desenvolve o backend.

Perguntas frequentes

Devo usar IHostedService ou BackgroundService para MQTT?

O BackgroundService é a escolha mais simples, porque oferece um único método ExecuteAsync com um token de parada. Ele implementa IHostedService por você, então ambos são registrados da mesma forma com AddHostedService.

Por que não injetar o DbContext diretamente no background service?

Hosted services são singletons e o DbContext tem escopo. Injetá-lo diretamente ou falha na validação ou mantém um único contexto vivo para sempre, então crie um escopo por lote com IServiceScopeFactory.

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

Use uma shared subscription do MQTT 5, como $share/devicehub/connectmq/+/telemetry. O broker então entrega cada mensagem a apenas um membro do grupo.

Posso publicar comandos para os dispositivos a partir de um controller?

Sim. Registre o IMqttClient conectado (ou um pequeno wrapper) como singleton e chame PublishAsync a partir do seu endpoint; não crie uma nova conexão por requisição HTTP.