MQTT in ASP.NET Core — BackgroundService, DI, Health Checks

To consume MQTT in ASP.NET Core, register a single MQTTnet IMqttClient in dependency injection, connect and reconnect it from a BackgroundService, push incoming messages into a Channel<T>, and process them in a second hosted service that creates a DI scope per message. A health check reports the connection state, and your API endpoints publish through the same client.

This tutorial builds that service step by step with MQTTnet 5 on .NET 8 or later. If you have not used MQTTnet before, read MQTT in C# with MQTTnet first; this article focuses on the hosting side.

The architecture

Component	Lifetime	Job
`IMqttClient`	Singleton	One connection per app instance, shared by everything
`MqttConnectionService`	Hosted service	Connects, reconnects, subscribes, disconnects on shutdown
`Channel<InboundMessage>`	Singleton	Bounded buffer between the network and your business logic
`TelemetryConsumer`	Hosted service	Reads the channel, creates a scope, calls your handler
`ITelemetryHandler`	Scoped	Your code — can use `DbContext` and other scoped services
`MqttHealthCheck`	Transient	Reports whether the client is connected

Splitting receiving from processing matters. The MQTTnet event handler runs on the client's dispatcher; if it waits on a database, every message behind it waits too, and keep-alive handling can suffer. The channel decouples the two and gives you natural backpressure.

Project setup and configuration

Create the project

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

Credentials live in user secrets locally and in environment variables (Mqtt__Username, Mqtt__Password) or a secret store in production — never in appsettings.json. Bind the section to a settings class:

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

The default client ID uses the machine name, which in Kubernetes is the pod name and therefore unique per replica. Two instances with the same client ID would keep kicking each other off the broker.

Register the services

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); // local function, see below
app.Run();

On MQTTnet 4, replace MqttClientFactory with MqttFactory and add using MQTTnet.Client; — the rest of the code is the same.

The connection service

This hosted service owns the connection. It reuses the reconnect loop from the MQTTnet tutorial and resubscribes in ConnectedAsync, so subscriptions survive every reconnect:

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;
            // Copy what we need; don't keep MQTTnet objects beyond the 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();
    }
}

Because the channel is bounded with FullMode = Wait, a slow consumer makes WriteAsync wait, which delays the PUBACK for QoS 1 messages. The broker then stops sending once the client's receive maximum is reached, instead of your process buffering messages until it runs out of memory. Tune the capacity to how many seconds of traffic you are willing to hold.

StopAsync sends a proper DISCONNECT when the host shuts down (Ctrl+C, a container stop, a deployment). Without it the broker only notices the missing client after the keep-alive expires and publishes its Last Will, if one is set.

The consumer and scoped handlers

If you inject a scoped service such as a DbContext straight into a hosted service, the app fails at startup with "Cannot consume scoped service 'AppDbContext' from singleton 'Microsoft.Extensions.Hosting.IHostedService'." The fix is a scope per message:

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; // save to a database, call another service, ...
    }
}

The try/catch around each message is essential: an unhandled exception in ExecuteAsync stops the hosted service — and since .NET 6 it stops the whole host by default. One malformed payload should produce a log entry, not an outage. If you need parallel processing, start several reader loops over the same channel and set SingleReader = false; keep in mind that ordering per topic is then no longer guaranteed.

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

Using context.Registration.FailureStatus lets the registration decide how serious a lost connection is; here it reports Degraded. The check is only part of the readiness endpoint, while /healthz/live runs no checks at all. That split is deliberate: if the broker has an outage, an orchestrator that restarts every pod on a failed liveness probe turns one problem into a reconnect storm. Readiness takes the instance out of load balancing while the reconnect loop does its job.

Publishing from an API endpoint

The singleton client can be injected anywhere. MQTTnet serializes packet writes internally, so concurrent requests can publish through the same connection:

Program.cs — after 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);
    });

Validating the ID is not cosmetic: a value containing +, # or / would change the topic structure. With MQTT 5, a QoS 1 publish to a topic without subscribers returns NoMatchingSubscribers, which is not an error. The message expiry keeps a command from being executed by a device that reconnects much later. Use the topic matcher to check that your command topics match what devices subscribe to.

Running several instances

With two replicas and a normal subscription, the broker delivers every message to both, and each one is processed twice. Use a shared subscription in the Topics setting — $share/telemetry-workers/testmqtt/aspnet/telemetry/# — and the broker load-balances messages across the group. Each replica still needs its own client ID. The details and the broker-specific caveats are in MQTT shared subscriptions.

Common pitfalls

Awaiting the connection during startup. If you connect in a constructor or block StartAsync until the broker answers, the app never starts while the broker is unreachable. Let ExecuteAsync connect in the background; the readiness check covers the gap.
Subscribing only after the first connect. With clean start the subscriptions disappear on every reconnect. Subscribe in ConnectedAsync.
Attaching event handlers more than once. If events are attached in code that runs on every reconnect, each message is processed once per attachment. Attach them exactly once.
Registering IMqttClient as scoped or transient. Every request would open its own broker connection, which is slow and quickly runs into connection limits. Keep it a singleton.

Test the service

dotnet run and check the log for MQTT connected and a SUBACK ... GrantedQoS1 line.
curl -i http://localhost:5000/healthz/ready should return 200 Healthy (use the port your app prints).
In the online MQTT client, publish {"celsius":21.5} to testmqtt/aspnet/telemetry/dev1; the handler logs it.
Subscribe to testmqtt/aspnet/devices/# in the browser, then curl -X POST http://localhost:5000/devices/dev1/commands -H 'Content-Type: application/json' -d '{"led":"on"}' and watch the command arrive.
Block the broker port (or disconnect Wi-Fi) for a moment: readiness switches to Degraded, then back to Healthy after the reconnect.

Send test telemetry to your ASP.NET Core service

If the connection fails with an exception you do not recognize, MQTTnet errors maps the common messages to causes.

With one singleton client, a connection service, a bounded channel and scoped handlers, MQTT becomes just another input to your ASP.NET Core app: configured through options, observable through health checks, shut down cleanly with the host, and ready to scale out with shared subscriptions.

Frequently asked questions

Why do I get 'Cannot consume scoped service from singleton' in my MQTT hosted service?

Hosted services are singletons, so they cannot take scoped services such as a DbContext in their constructor. Inject IServiceScopeFactory instead and create a scope for each message (or batch) you process.

Should the MQTT health check fail the liveness probe?

Usually not. If the broker is down, restarting your app will not fix it and can cause a reconnect storm. Report the MQTT connection on a readiness or status endpoint and keep liveness for problems a restart actually solves.

How do I run several instances without processing every message twice?

Give each instance a unique client ID and subscribe with a shared subscription such as $share/telemetry-workers/testmqtt/aspnet/telemetry/#. The broker then distributes messages across the group instead of sending a copy to every instance.

MQTT in ASP.NET Core: A Hosted Service Consumer with DI and Health Checks