MQTT con ASP.NET Core: un servicio en segundo plano que guarda y transmite datos de dispositivos

Una integración MQTT con ASP.NET Core es la columna vertebral natural de un backend IoT en .NET: los dispositivos publican en un broker y tu aplicación web se suscribe, almacena y sirve los datos. En esta guía alojarás un suscriptor MQTTnet como BackgroundService, escribirás cada lectura en una base de datos con EF Core y la enviarás a los navegadores en tiempo real con SignalR. La aplicación terminada se ejecuta contra broker.connectmq.com para las pruebas y pasa a un broker privado de ConnectMQ solo con cambios de configuración.

Arquitectura de un servicio MQTT con ASP.NET Core

La forma de la solución es sencilla y cada pieza tiene una única tarea:

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

¿Por qué un BackgroundService y no un controlador? MQTT es una conexión de larga duración basada en push. Un hosted service arranca con la aplicación, mantiene una conexión abierta durante toda su vida y se detiene de forma ordenada al apagarse, que es exactamente el ciclo de vida que quieres. Los controladores tienen ámbito de solicitud y abrirían y cerrarían conexiones constantemente.

Dos reglas de diseño lo mantienen robusto:

1. El cliente MQTT es un singleton propiedad del hosted service. Nunca crees un cliente por solicitud.
2. Los contextos de base de datos tienen ámbito (scoped). El hosted service es un singleton, así que debe crear un scope por lote de mensajes con IServiceScopeFactory.

Configuración del proyecto

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

SignalR forma parte del framework compartido de ASP.NET Core, así que no hace falta ningún paquete de servidor adicional.

Añade la configuración MQTT a appsettings.json. Para el servidor MQTT público gratuito, usa cualquier client ID único y un prefijo de topic aleatorio; el área pública la puede leer cualquiera y sirve solo para pruebas y prototipos, así que no envíes nada sensible ni datos de producción.

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

El modelo de datos y el 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; } = "";
}

El hub de SignalR

El hub en sí puede estar vacío. El servicio en segundo plano envía a través de IHubContext<TelemetryHub> y los navegadores solo escuchan.

using Microsoft.AspNetCore.SignalR;

public class TelemetryHub : Hub { }

El servicio MQTT en segundo plano

Este es el corazón de la integración MQTT con ASP.NET Core. Se conecta, se suscribe, se reconecta con backoff y procesa los mensajes sin bloquear el bucle de recepción 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();
        }
    }
}

Algunas decisiones que conviene explicar:

• El handler de mensajes solo analiza y encola. Las llamadas a la base de datos y a SignalR ocurren en un bucle aparte. Si el handler esperara a una base de datos lenta, MQTTnet dejaría de leer el socket, los keep-alive podrían agotar su tiempo y el broker cerraría la conexión.
• Un canal acotado con DropOldest limita la memoria si la base de datos se cae. Para la telemetría, perder las lecturas más antiguas suele ser mejor que tumbar el proceso. Si no puedes perder datos, cambia a Wait y acepta la contrapresión.
• El procesamiento por lotes convierte cientos de inserciones en una sola llamada a SaveChangesAsync.
• La sesión persistente (CleanStart(false) más una expiración de sesión de 30 minutos) significa que los mensajes QoS 1 publicados mientras tu aplicación se reinicia se entregan cuando vuelve a conectarse, y la suscripción se conserva. Eso solo funciona si el client ID permanece estable, y por eso vive en la configuración.
• Los payloads mal formados se registran y se descartan, nunca se relanzan. Un dispositivo defectuoso no debe tumbar la ingesta.

Conexión de todo en 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 sirve para una demo; en un proyecto real usa las migraciones de EF Core.

Un panel en vivo mínimo

Crea wwwroot/index.html con una lista y el cliente JavaScript de SignalR desde un CDN. El script esencial es:

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

Fíjate en los nombres de propiedad en camelCase: el protocolo JSON de SignalR serializa por defecto las propiedades de C# en camelCase.

Prueba la canalización de extremo a extremo

Ejecuta la aplicación con dotnet run, abre la página y publica una lectura simulada desde una 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}'

La lectura debería aparecer en el navegador en una fracción de segundo y poder consultarse en /api/readings/boiler-1. Para más variantes de línea de comandos, consulta los ejemplos de mosquitto_pub y mosquitto_sub.

Envío de comandos a los dispositivos

La ingesta es solo la mitad de la mayoría de los backends IoT; la otra mitad es el control. Un error habitual es abrir una nueva conexión MQTT dentro de un endpoint HTTP para cada comando. Es lento, desperdicia conexiones del broker (que cuentan para la cuota de tu plan) y falla bajo carga. En su lugar, mantén un segundo cliente de larga duración para publicar, o expón el cliente de ingesta mediante un pequeño 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);
}

El intervalo de expiración del mensaje importa en los comandos: si un dispositivo está desconectado diez minutos, casi nunca querrás que ejecute un comando de «abrir válvula» en cuanto se reconecte. Con una expiración de 30 segundos, el broker descarta el comando obsoleto en lugar de entregarlo. Comprueba el reason code del MqttClientPublishResult devuelto y comunica los fallos a quien hizo la llamada, en lugar de devolver 200 OK por un comando que nunca se aceptó.

Compartir un cliente entre la ingesta y los comandos o usar dos es una cuestión de criterio. Dos clientes dan logs más claros y un comportamiento de reconexión independiente; uno ahorra una conexión. En un plan pequeño, uno suele bastar.

Observabilidad

Un servicio en segundo plano falla en silencio a menos que lo hagas ruidoso. Como mínimo, registra los cambios de estado de la conexión con su motivo, cuenta los mensajes recibidos y descartados por minuto y registra la profundidad de la cola del canal. El ILogger integrado de ASP.NET Core y un health check que informe de IsConnected cubren la mayoría de las necesidades; si ya usas OpenTelemetry, expón esos contadores como métricas para poder alertar cuando la ingesta se detenga mientras los dispositivos siguen publicando.

Lista de comprobación para producción

Las suscripciones compartidas y QoS 2 están disponibles en el plan Serverless de pago y superiores, así que tenlo en cuenta antes de ejecutar más de una instancia de ingesta. La guía de funciones de MQTT 5 explica con más detalle las suscripciones compartidas y la expiración de sesión.

Paso a un broker privado

Los dispositivos de producción deben hablar con un broker privado y autenticado. ConnectMQ ofrece un broker MQTT gestionado con TLS en el puerto 8883, MQTT 3.1.1 y 5.0 y cuotas por plan, de modo que tu aplicación ASP.NET Core solo tiene que cambiar la configuración:

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 ConnectMQ vincula los client ID a las credenciales emitidas, usa para el backend el client ID generado en el portal y da a cada dispositivo sus propias credenciales. Si quieres entender las opciones del cliente con más profundidad, el tutorial de MQTTnet cubre TLS, las propiedades de MQTT 5 y los detalles de reconexión, y la guía de bibliotecas cliente MQTT compara .NET con otros ecosistemas. Si todavía estás evaluando dónde alojar el broker, lee sobre cómo ejecutar un broker MQTT en la nube.

Empieza con ConnectMQ

Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, pon las credenciales generadas en user secrets y tu servicio MQTT con ASP.NET Core tendrá forma de producción desde el primer día. Usa el explorador de topics en vivo del portal para observar el tráfico de los dispositivos mientras desarrollas el backend.

Preguntas frecuentes

¿Debo usar IHostedService o BackgroundService para MQTT?

BackgroundService es la opción más sencilla porque te da un único método ExecuteAsync con un token de parada. Implementa IHostedService por ti, así que ambos se registran igual con AddHostedService.

¿Por qué no inyectar DbContext directamente en el servicio en segundo plano?

Los hosted services son singletons y DbContext tiene ámbito scoped. Inyectarlo directamente o bien falla la validación o bien mantiene un contexto vivo para siempre, así que crea un scope por lote con IServiceScopeFactory.

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

Usa una suscripción compartida de MQTT 5 como $share/devicehub/connectmq/+/telemetry. El broker entrega entonces cada mensaje a un solo miembro del grupo.

¿Puedo publicar comandos a los dispositivos desde un controlador?

Sí. Registra el IMqttClient conectado (o un pequeño wrapper) como singleton y llama a PublishAsync desde tu endpoint; no crees una nueva conexión por cada solicitud HTTP.