ASP.NET Core'da MQTT: BackgroundService, DI, Health Check

ASP.NET Core'da MQTT tüketmek için dependency injection'a tek bir MQTTnet IMqttClient kaydedin, onu bir BackgroundService içinden bağlayıp yeniden bağlayın, gelen mesajları bir Channel<T>'ye aktarın ve her mesaj için bir DI scope'u oluşturan ikinci bir hosted service'te işleyin. Bir health check bağlantı durumunu raporlar, API endpoint'leriniz de aynı istemci üzerinden publish eder.

Bu rehber bu servisi .NET 8 veya üzeri üzerinde MQTTnet 5 ile adım adım kuruyor. MQTTnet'i daha önce kullanmadıysanız önce C# ve MQTTnet ile MQTT yazısını okuyun; bu yazı hosting tarafına odaklanıyor.

Mimari

Bileşen	Yaşam süresi	Görev
`IMqttClient`	Singleton	Uygulama instance'ı başına tek bağlantı, herkes tarafından paylaşılır
`MqttConnectionService`	Hosted service	Bağlanır, yeniden bağlanır, abone olur, kapanışta bağlantıyı keser
`Channel<InboundMessage>`	Singleton	Ağ ile iş mantığınız arasında sınırlı bir tampon
`TelemetryConsumer`	Hosted service	Kanalı okur, scope oluşturur, handler'ınızı çağırır
`ITelemetryHandler`	Scoped	Sizin kodunuz — `DbContext` ve diğer scoped servisleri kullanabilir
`MqttHealthCheck`	Transient	İstemcinin bağlı olup olmadığını raporlar

Mesaj almayı işlemeden ayırmak önemlidir. MQTTnet event handler'ı istemcinin dispatcher'ında çalışır; bir veritabanını beklerse arkasındaki tüm mesajlar da bekler ve keep-alive yönetimi bundan etkilenebilir. Kanal ikisini birbirinden ayırır ve doğal bir backpressure sağlar.

Proje kurulumu ve yapılandırma

Projeyi oluşturun

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

Kimlik bilgileri yerelde user secrets'ta, production'da ise ortam değişkenlerinde (Mqtt__Username, Mqtt__Password) veya bir secret store'da durur — asla appsettings.json'da değil. Bölümü bir ayar sınıfına bağlayın:

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

Varsayılan client ID makine adını kullanır; Kubernetes'te bu pod adıdır ve dolayısıyla her replica için benzersizdir. Aynı client ID'ye sahip iki instance birbirini sürekli broker'dan düşürür.

Servisleri kaydetme

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); // yerel fonksiyon, aşağıya bakın
app.Run();

MQTTnet 4'te MqttClientFactory yerine MqttFactory kullanın ve using MQTTnet.Client; ekleyin; kodun geri kalanı aynıdır.

Bağlantı servisi

Bu hosted service bağlantının sahibidir. MQTTnet rehberindeki reconnect döngüsünü yeniden kullanır ve ConnectedAsync içinde yeniden abone olur; böylece abonelikler her yeniden bağlanmadan sonra korunur:

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;
            // Gerekeni kopyalıyoruz; MQTTnet nesnelerini handler dışında tutmuyoruz.
            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();
    }
}

Kanal FullMode = Wait ile sınırlandırıldığı için yavaş bir tüketici WriteAsync'i bekletir; bu da QoS 1 mesajlarının PUBACK'ini geciktirir. Böylece süreciniz belleği bitene kadar mesaj biriktirmek yerine, istemcinin receive maximum değerine ulaşıldığında broker göndermeyi durdurur. Kapasiteyi, kaç saniyelik trafiği tutmaya razı olduğunuza göre ayarlayın.

StopAsync, host kapanırken (Ctrl+C, container durdurma, bir deployment) düzgün bir DISCONNECT gönderir. Bu olmazsa broker istemcinin yokluğunu ancak keep-alive süresi dolduğunda fark eder ve tanımlıysa Last Will mesajını yayınlar.

Tüketici ve scoped handler'lar

DbContext gibi scoped bir servisi doğrudan bir hosted service'e inject ederseniz uygulama açılışta "Cannot consume scoped service 'AppDbContext' from singleton 'Microsoft.Extensions.Hosting.IHostedService'." hatasıyla durur. Çözüm, mesaj başına bir scope'tur:

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; // veritabanına kaydedin, başka bir servisi çağırın, ...
    }
}

Her mesajın etrafındaki try/catch şarttır: ExecuteAsync içinde yakalanmayan bir exception hosted service'i durdurur — ve .NET 6'dan beri varsayılan olarak tüm host'u. Bozuk bir payload bir kesintiye değil, bir log kaydına yol açmalıdır. Paralel işleme gerekiyorsa aynı kanal üzerinde birkaç okuma döngüsü başlatın ve SingleReader = false ayarlayın; bu durumda topic başına sıranın artık garanti edilmediğini unutmayın.

Health check'ler

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

context.Registration.FailureStatus kullanmak, kopan bağlantının ne kadar ciddi olduğuna kaydın karar vermesini sağlar; burada Degraded raporlanır. Check yalnızca readiness endpoint'inin parçasıdır; /healthz/live ise hiçbir check çalıştırmaz. Bu ayrım bilinçlidir: broker çöktüğünde, başarısız liveness probe'unda her pod'u yeniden başlatan bir orchestrator tek bir sorunu bir reconnect fırtınasına dönüştürür. Readiness, reconnect döngüsü işini yaparken instance'ı load balancing dışına alır.

Bir API endpoint'inden publish

Singleton istemci her yere inject edilebilir. MQTTnet paket yazımlarını dahili olarak sıraya koyar; bu yüzden eşzamanlı istekler aynı bağlantı üzerinden publish edebilir:

Program.cs — app.Run() sonrası

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

ID'yi doğrulamak süs değildir: +, # veya / içeren bir değer topic yapısını değiştirir. MQTT 5'te abonesi olmayan bir topic'e yapılan QoS 1 publish NoMatchingSubscribers döndürür; bu bir hata değildir. Message expiry, çok sonra yeniden bağlanan bir cihazın komutu yine de çalıştırmasını engeller. Komut topic'lerinizin cihazların abone olduğu filtrelerle eşleştiğini kontrol etmek için topic matcher aracını kullanın.

Birden fazla instance çalıştırma

İki replica ve normal bir abonelikle broker her mesajı ikisine de teslim eder ve her mesaj iki kez işlenir. Topics ayarında bir shared subscription kullanın — $share/telemetry-workers/testmqtt/aspnet/telemetry/# — ve broker mesajları grup içinde dağıtsın. Her replica'nın yine kendi client ID'si olmalıdır. Ayrıntılar ve broker'a özgü noktalar MQTT shared subscription yazısında.

Sık yapılan hatalar

Açılışta bağlantıyı beklemek. Constructor'da bağlanırsanız veya broker cevap verene kadar StartAsync'i bloklarsanız, broker erişilemezken uygulama hiç açılmaz. ExecuteAsync'in arka planda bağlanmasına izin verin; aradaki süreyi readiness check karşılar.
Yalnızca ilk bağlantıdan sonra abone olmak. Clean start ile abonelikler her yeniden bağlanmada kaybolur. ConnectedAsync içinde abone olun.
Event handler'ları birden fazla kez eklemek. Event'ler her yeniden bağlanmada çalışan bir kodda eklenirse her mesaj, ekleme sayısı kadar işlenir. Onları tam olarak bir kez ekleyin.
IMqttClient'ı scoped veya transient kaydetmek. Her istek kendi broker bağlantısını açar; bu hem yavaştır hem de hızla bağlantı limitlerine takılır. Singleton olarak tutun.

Servisi test edin

dotnet run çalıştırın ve log'da MQTT connected ile bir SUBACK ... GrantedQoS1 satırı arayın.
curl -i http://localhost:5000/healthz/ready komutu 200 Healthy döndürmelidir (uygulamanızın yazdırdığı portu kullanın).
Online MQTT istemcisinde testmqtt/aspnet/telemetry/dev1 topic'ine {"celsius":21.5} publish edin; handler bunu loglar.
Tarayıcıda testmqtt/aspnet/devices/# topic'ine abone olun, ardından curl -X POST http://localhost:5000/devices/dev1/commands -H 'Content-Type: application/json' -d '{"led":"on"}' çalıştırın ve komutun geldiğini izleyin.
Broker portunu kısa bir süre engelleyin (veya Wi-Fi'ı kesin): readiness Degraded durumuna geçer, yeniden bağlandıktan sonra Healthy'ye döner.

ASP.NET Core servisinize test telemetrisi gönderin

Bağlantı tanımadığınız bir exception ile başarısız olursa, MQTTnet hataları yazısı yaygın mesajları nedenleriyle eşleştirir.

Tek bir singleton istemci, bir bağlantı servisi, sınırlı bir kanal ve scoped handler'larla MQTT, ASP.NET Core uygulamanızın sıradan bir girdisine dönüşür: options ile yapılandırılır, health check'lerle izlenir, host ile birlikte düzgünce kapanır ve shared subscription'larla ölçeklenmeye hazırdır.

Sıkça sorulan sorular

MQTT hosted service'imde neden 'Cannot consume scoped service from singleton' hatası alıyorum?

Hosted service'ler singleton'dır; bu yüzden constructor'larında DbContext gibi scoped servisleri alamazlar. Bunun yerine IServiceScopeFactory inject edin ve işlediğiniz her mesaj (veya batch) için bir scope oluşturun.

MQTT health check'i liveness probe'u düşürmeli mi?

Genellikle hayır. Broker çökmüşse uygulamanızı yeniden başlatmak sorunu çözmez ve bir reconnect fırtınasına yol açabilir. MQTT bağlantısını readiness veya durum endpoint'inde raporlayın, liveness'ı yeniden başlatmanın gerçekten çözdüğü sorunlara bırakın.

Her mesajı iki kez işlemeden birden fazla instance nasıl çalıştırılır?

Her instance'a benzersiz bir client ID verin ve $share/telemetry-workers/testmqtt/aspnet/telemetry/# gibi bir shared subscription ile abone olun. Broker böylece her instance'a bir kopya göndermek yerine mesajları grup içinde dağıtır.

ASP.NET Core'da MQTT: DI ve Health Check ile Hosted Service Tüketici