MQTT en C# con MQTTnet 4 y 5: conectar, suscribir, reconectar

MQTTnet es la librería MQTT estándar para .NET. Para crear un cliente en C#, obtén un IMqttClient de la factory, describe la conexión con MqttClientOptionsBuilder, añade un handler ApplicationMessageReceivedAsync y llama a ConnectAsync, SubscribeAsync y PublishAsync. Desde MQTTnet 5 ya no existe el managed client, así que la reconexión la escribes tú: son unas veinte líneas.

Este tutorial usa MQTTnet 5 y señala cada punto en el que la versión 4 es distinta, para que puedas seguirlo con cualquiera de las dos. Todos los ejemplos se conectan a broker.connectmq.com y usan topics bajo testmqtt/dotnet/.

Instalar y elegir versión

dotnet new console -n MqttDemo && cd MqttDemo
dotnet add package MQTTnet

Esto instala la última versión mayor. Si mantienes un código existente con MQTTnet 4, fíjala con --version 4.* hasta que estés listo para migrar. Las diferencias con las que te toparás primero:

ConvertPayloadToString() funciona en ambas, por eso los ejemplos lo usan.

Conectar

using MQTTnet;            // v4: añade también `using MQTTnet.Client;`
using MQTTnet.Formatter;
using MQTTnet.Protocol;

var factory = new MqttClientFactory();   // v4: new MqttFactory()
using var client = factory.CreateMqttClient();

var options = new MqttClientOptionsBuilder()
    .WithTcpServer("broker.connectmq.com", 1883)
    .WithClientId($"dotnet-{Guid.NewGuid():N}"[..20])
    .WithCredentials(Environment.GetEnvironmentVariable("MQTT_USER"),
                     Environment.GetEnvironmentVariable("MQTT_PASS"))
    .WithProtocolVersion(MqttProtocolVersion.V500)
    .WithKeepAlivePeriod(TimeSpan.FromSeconds(30))
    .WithCleanStart()
    .Build();

var result = await client.ConnectAsync(options, CancellationToken.None);
Console.WriteLine($"Conectado: {result.ResultCode}");

En el endpoint compartido de pruebas puedes omitir WithCredentials; en tu propio broker pasa el usuario y la contraseña desde variables de entorno en lugar de escribirlos en el código. Si el broker rechaza la conexión, ConnectAsync lanza MqttClientConnectingFailedException; su ResultCode (por ejemplo BadUserNameOrPassword o NotAuthorized) te dice por qué. La referencia de reason codes MQTT lista todos los valores, y las excepciones más comunes se explican en errores de MQTTnet.

Añadir un Last Will

La mayoría de clientes deberían avisar cuando desaparecen. Añade un mensaje will al mismo builder de opciones:

    .WithWillTopic("testmqtt/dotnet/status")
    .WithWillPayload("offline")
    .WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
    .WithWillRetain()

El broker publica el will solo cuando la conexión termina sin un paquete DISCONNECT: un crash, una pérdida de red o un timeout de keep-alive. Un DisconnectAsync() limpio lo suprime, así que publica tú mismo un estado retenido online tras conectar y un offline antes de un apagado planificado. Con MQTT 5, .WithWillDelayInterval(10) da al cliente unos segundos para volver antes de que salga el will, lo que evita estados que parpadean durante cortes breves. MQTT Last Will cubre los detalles.

Suscribirse y procesar mensajes

Registra el handler antes de suscribirte; si no, los mensajes retenidos que llegan justo después del SUBACK pueden aparecer cuando aún nadie escucha:

client.ApplicationMessageReceivedAsync += e =>
{
    var msg = e.ApplicationMessage;
    Console.WriteLine($"{msg.Topic} (QoS {(int)msg.QualityOfServiceLevel}): {msg.ConvertPayloadToString()}");
    return Task.CompletedTask;
};

var subscribe = factory.CreateSubscribeOptionsBuilder()
    .WithTopicFilter(f => f.WithTopic("testmqtt/dotnet/sensors/#")
                           .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
    .Build();

var subResult = await client.SubscribeAsync(subscribe, CancellationToken.None);
foreach (var item in subResult.Items)
    Console.WriteLine($"{item.TopicFilter.Topic}: {item.ResultCode}");

Imprime siempre el resultado del subscribe. Un broker que deniega una suscripción por su ACL no cierra la conexión: simplemente devuelve un código de fallo como NotAuthorized en el SUBACK, y tu handler nunca se dispara. Revisar ResultCode convierte un fallo silencioso en una línea de log.

El handler se ejecuta en el dispatcher interno de MQTTnet. Mantenlo corto: parsea, delega el trabajo (a un Channel<T>, una cola o una tarea en segundo plano) y retorna. Un trabajo largo dentro del handler retrasa los acknowledgements y todos los mensajes que vienen detrás.

Publicar

using System.Text.Json;

var payload = JsonSerializer.Serialize(new { celsius = 21.5, ts = DateTimeOffset.UtcNow });

var message = new MqttApplicationMessageBuilder()
    .WithTopic("testmqtt/dotnet/sensors/lab")
    .WithPayload(payload)
    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
    .WithRetainFlag(false)
    .WithContentType("application/json")      // solo MQTT 5
    .WithUserProperty("source", "mqttnet-demo") // solo MQTT 5
    .Build();

var pub = await client.PublishAsync(message, CancellationToken.None);
Console.WriteLine($"Publish: {pub.ReasonCode}");

Con QoS 1, PublishAsync termina cuando llega el PUBACK, así que el reason code es la respuesta del broker. Con QoS 0 termina en cuanto el paquete se escribe en el socket. El broker puede rechazar payloads grandes por su tamaño máximo de paquete; la calculadora de tamaño de paquete muestra cuánto ocupa realmente un mensaje en la red.

TLS y WebSockets

Cambia el transporte en el builder de opciones; todo lo demás sigue igual:

var tlsOptions = new MqttClientOptionsBuilder()
    .WithTcpServer("broker.connectmq.com", 8883)
    .WithTlsOptions(o => o.UseTls())
    .WithCredentials(Environment.GetEnvironmentVariable("MQTT_USER"),
                     Environment.GetEnvironmentVariable("MQTT_PASS"))
    .Build();

var wssOptions = new MqttClientOptionsBuilder()
    .WithWebSocketServer(o => o.WithUri("wss://broker.connectmq.com:8084/mqtt"))
    .WithTlsOptions(o => o.UseTls())
    .Build();

El certificado del broker se valida contra el almacén de confianza del sistema operativo. Resiste la tentación de añadir WithCertificateValidationHandler(_ => true) para que desaparezca un error de handshake; busca la causa, empezando por probar la conexión TLS desde una terminal. Las versiones antiguas de MQTTnet 4 usaban .WithTls(), que ahora está obsoleto.

Reconectar sin ManagedMqttClient

MQTTnet 4 ofrecía ManagedMqttClient, que reconectaba y encolaba mensajes por ti. En la versión 5 desapareció, y los ejemplos oficiales usan un bucle sencillo en su lugar. Esta es una versión completa:

client.ConnectedAsync += async e =>
{
    // Con clean start el broker olvidó nuestras suscripciones; con sesión persistente quizá no.
    if (!e.ConnectResult.IsSessionPresent)
        await client.SubscribeAsync(subscribe, CancellationToken.None);
    Console.WriteLine("Conectado, sesión presente: " + e.ConnectResult.IsSessionPresent);
};

client.DisconnectedAsync += e =>
{
    Console.WriteLine($"Desconectado ({e.Reason}): {e.Exception?.Message}");
    return Task.CompletedTask;
};

using var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, a) => { a.Cancel = true; cts.Cancel(); };

var delay = TimeSpan.FromSeconds(1);
while (!cts.IsCancellationRequested)
{
    try
    {
        if (!await client.TryPingAsync(cts.Token))
        {
            await client.ConnectAsync(options, cts.Token);
            delay = TimeSpan.FromSeconds(1);           // reinicia el backoff tras conectar
        }
    }
    catch (Exception ex) when (ex is not OperationCanceledException)
    {
        Console.WriteLine($"Fallo al reconectar: {ex.Message}");
        delay = TimeSpan.FromSeconds(Math.Min(delay.TotalSeconds * 2, 60));
    }

    try { await Task.Delay(delay + TimeSpan.FromMilliseconds(Random.Shared.Next(500)), cts.Token); }
    catch (OperationCanceledException) { break; }
}

await client.DisconnectAsync();

Tres detalles lo hacen robusto. La suscripción vive en ConnectedAsync, así que se restaura tras cada reconexión, no solo la primera. La espera crece exponencialmente con un poco de jitter aleatorio, para que mil clientes que pierden el broker a la vez no reconecten todos en el mismo segundo; las buenas prácticas de reconexión explican por qué importa. Y TryPingAsync sirve además de health check: devuelve false tanto si el cliente no está conectado como si la conexión está medio abierta.

Buffer mientras estás offline

El managed client también encolaba los mensajes salientes. Si lo necesitas, pon un Channel<MqttApplicationMessage> entre tu código y el cliente: los productores escriben en el canal y una única tarea consumidora lee de él y llama a PublishAsync solo cuando client.IsConnected es true, reintentando el mensaje si la publicación lanza una excepción. Limita el canal (Channel.CreateBounded) para que un corte largo no agote la memoria, y decide explícitamente si descartas los mensajes más antiguos o los más nuevos cuando se llene.

Funciones de MQTT 5 que vale la pena usar

Con WithProtocolVersion(MqttProtocolVersion.V500) activado, algunas opciones compensan de inmediato:

• Session expiry: .WithSessionExpiryInterval(3600) junto con .WithCleanStart(false) conserva suscripciones y mensajes QoS 1 en cola durante una hora tras una desconexión. Pruébalo contra un broker privado; los endpoints compartidos de pruebas normalmente no guardan sesiones offline.
• Reason codes y reason strings: los resultados de connect, subscribe y publish incluyen ResultCode/ReasonCode y a menudo un ReasonString. Regístralos; convierten "no funciona" en "cuota excedida".
• User properties: metadatos clave/valor como un trace ID o la versión de firmware, sin cambiar el formato del payload.
• Message expiry: .WithMessageExpiryInterval(60) en un mensaje evita que se entreguen comandos caducados a un dispositivo que reconecta una hora después.
• Request/response: .WithResponseTopic() y .WithCorrelationData() ofrecen una forma estándar de implementar RPC sobre MQTT.

Pruébalo

Ejecuta la app de consola, abre el cliente MQTT online y suscríbete a testmqtt/dotnet/#. Deberías ver llegar la lectura JSON en testmqtt/dotnet/sensors/lab. Publica desde el navegador en testmqtt/dotnet/sensors/test y el handler de C# lo imprimirá. Para probar el bucle de reconexión, desconecta la red unos segundos y observa cómo el log pasa de Desconectado a Conectado, sesión presente: False seguido de un SUBACK nuevo.

Mira en directo los mensajes de tu cliente MQTTnet

Si estás construyendo una web API o un worker service, el siguiente paso es mover este cliente a un hosted service con inyección de dependencias y health checks: consulta MQTT en ASP.NET Core.

MQTTnet te da una API pequeña y explícita: un builder de opciones, tres llamadas async y un puñado de eventos. Añade un bucle de reconexión que se vuelva a suscribir en ConnectedAsync, registra cada código de resultado, y el mismo cliente funcionará desde una prueba de consola en el endpoint compartido hasta un servicio de producción en tu propio broker.

Preguntas frecuentes