ASP.NET Core MQTT: Cihaz Verisini Saklayan ve Canlı Aktaran Bir Background Service

ASP.NET Core MQTT entegrasyonu, bir .NET IoT backend'inin doğal omurgasıdır: cihazlar bir broker'a yayın yapar, web uygulamanız ise abone olur, veriyi saklar ve sunar. Bu rehberde bir MQTTnet subscriber'ı BackgroundService olarak barındıracak, her okumayı EF Core ile bir veritabanına yazacak ve SignalR ile tarayıcılara gerçek zamanlı ileteceksiniz. Bitmiş uygulama test için broker.connectmq.com üzerinde çalışır ve yalnızca yapılandırma değişikliğiyle özel bir ConnectMQ broker'a geçer.

ASP.NET Core MQTT Servisinin Mimarisi

Çözümün yapısı basittir ve her parçanın tek bir görevi vardır:

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

Neden controller değil de BackgroundService? MQTT uzun ömürlü, push tabanlı bir bağlantıdır. Bir hosted service uygulamayla birlikte başlar, ömrü boyunca tek bir bağlantıyı açık tutar ve kapanışta düzgünce durur; bu tam olarak istediğiniz yaşam döngüsüdür. Controller'lar istek kapsamlıdır ve sürekli bağlantı açıp kapatırlar.

İki tasarım kuralı bu yapıyı sağlam tutar:

1. MQTT client, hosted service'in sahip olduğu bir singleton'dır. Asla istek başına client oluşturmayın.
2. Veritabanı context'leri scoped'dur. Hosted service bir singleton olduğu için her mesaj grubu için IServiceScopeFactory ile bir scope oluşturmalıdır.

Proje Kurulumu

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

SignalR, ASP.NET Core paylaşılan framework'ünün bir parçasıdır; bu nedenle ek bir sunucu paketi gerekmez.

MQTT ayarlarını appsettings.json dosyasına ekleyin. Herkese açık broker için benzersiz herhangi bir client ID ve rastgele bir topic öneki kullanın; herkese açık alan herkes tarafından okunabildiği için hassas hiçbir şey göndermeyin.

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

Veri Modeli ve 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; } = "";
}

SignalR Hub

Hub'ın kendisi boş olabilir. Background service IHubContext<TelemetryHub> üzerinden gönderim yapar, tarayıcılar ise yalnızca dinler.

using Microsoft.AspNetCore.SignalR;

public class TelemetryHub : Hub { }

MQTT Background Service

ASP.NET Core MQTT entegrasyonunun kalbi burasıdır. Servis bağlanır, abone olur, geri çekilmeyle yeniden bağlanır ve MQTTnet'in alım döngüsünü bloklamadan mesajları işler.

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

Açıklamaya değer birkaç tercih:

• Mesaj handler'ı yalnızca ayrıştırır ve kuyruğa ekler. Veritabanı ve SignalR çağrıları ayrı bir döngüde gerçekleşir. Handler yavaş bir veritabanını await etseydi MQTTnet soketi okumayı bırakır, keep-alive'lar zaman aşımına uğrayabilir ve broker bağlantıyı düşürürdü.
• DropOldest ile sınırlı bir channel, veritabanı çökerse bellek kullanımını sınırlar. Telemetri için en eski okumaları kaybetmek genellikle süreci çökertmekten iyidir. Veri kaybedemiyorsanız Wait moduna geçin ve geri basıncı kabul edin.
• Toplu işleme (batching), yüzlerce ekleme işlemini tek bir SaveChangesAsync çağrısına dönüştürür.
• Kalıcı oturum (CleanStart(false) ve 30 dakikalık oturum sona erme süresi), uygulamanız yeniden başlarken yayınlanan QoS 1 mesajlarının yeniden bağlandığında teslim edilmesi ve aboneliğin korunması anlamına gelir. Bu yalnızca client ID sabit kaldığında çalışır; bu yüzden ID yapılandırmada tutulur.
• Hatalı biçimli payload'lar loglanır ve atılır, asla yeniden fırlatılmaz. Tek bir sorunlu cihaz veri alımını durdurmamalıdır.

Program.cs'de Bağlantıları Kurmak

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 bir demo için yeterlidir; gerçek bir projede EF Core migration'larını kullanın.

Minimal Bir Canlı Pano

Bir liste ve CDN'den yüklenen SignalR JavaScript istemcisiyle wwwroot/index.html dosyasını oluşturun. Temel betik şudur:

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

camelCase özellik adlarına dikkat edin: SignalR'ın JSON protokolü C# özelliklerini varsayılan olarak camelCase biçiminde serileştirir.

Hattı Uçtan Uca Test Edin

Uygulamayı dotnet run ile çalıştırın, sayfayı açın ve ardından bir terminalden sahte bir okuma yayınlayın:

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}'

Okuma saniyenin kesri içinde tarayıcıda görünmeli ve /api/readings/boiler-1 adresinden alınabilmelidir. Daha fazla komut satırı varyasyonu için mosquitto_pub ve mosquitto_sub örneklerine bakın.

Cihazlara Geri Komut Göndermek

Veri alımı çoğu IoT backend'inin yalnızca yarısıdır; diğer yarısı kontroldür. Yaygın bir hata, her komut için bir HTTP uç noktası içinde yeni bir MQTT bağlantısı açmaktır. Bu yavaştır, broker bağlantılarını (planınızın kotasından düşülür) boşa harcar ve yük altında başarısız olur. Bunun yerine yayın için ikinci bir uzun ömürlü client tutun veya veri alım client'ını küçük bir singleton sarmalayıcı aracılığıyla sunun:

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

Mesaj sona erme aralığı komutlar için önemlidir: bir cihaz on dakika çevrimdışı kalırsa, yeniden bağlandığı anda bir "vanayı aç" komutunu yürütmesini neredeyse hiçbir zaman istemezsiniz. 30 saniyelik bir sona erme süresiyle broker bayat komutu teslim etmek yerine atar. Döndürülen MqttClientPublishResult içindeki reason code'u kontrol edin ve hiç kabul edilmemiş bir komut için 200 OK döndürmek yerine hataları çağırana bildirin.

Veri alımı ve komutlar arasında tek bir client paylaşmak ya da iki client kullanmak bir değerlendirme meselesidir. İki client daha net loglar ve bağımsız yeniden bağlanma davranışı sağlar; tek client ise bir bağlantı tasarrufu sağlar. Küçük bir planda genellikle tek client yeterlidir.

Gözlemlenebilirlik

Bir background service, siz onu görünür kılmadıkça sessizce başarısız olur. En azından bağlantı durumu değişikliklerini nedeniyle birlikte loglayın, dakika başına alınan ve atılan mesajları sayın ve channel'ın kuyruk derinliğini kaydedin. ASP.NET Core'un yerleşik ILogger arayüzü ve IsConnected durumunu bildiren bir health check ihtiyaçların çoğunu karşılar; zaten OpenTelemetry kullanıyorsanız, cihazlar yayın yapmaya devam ederken veri alımı durduğunda uyarı verebilmek için bu sayaçları metrik olarak sunun.

Üretim Ortamı Kontrol Listesi

Shared subscription'lar ve QoS 2, ücretli Serverless plan ve üzerinde mevcuttur; bu yüzden birden fazla veri alım örneği çalıştırmadan önce bunu planlayın. MQTT 5 özellikleri rehberi, shared subscription'ları ve oturum sona erme süresini daha ayrıntılı açıklar.

Özel Bir Broker'a Geçiş

Üretim cihazları özel ve kimlik doğrulamalı bir broker ile konuşmalıdır. ConnectMQ, 8883 portunda TLS, MQTT 3.1.1 ve 5.0 ile plan bazlı kotalar sunan tam yönetilen bir MQTT hizmeti sağlar; böylece ASP.NET Core uygulamanızın yalnızca yapılandırmayı değiştirmesi gerekir:

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"

ConnectMQ client ID'leri verilen kimlik bilgilerine bağladığı için backend'de portalda üretilen client ID'yi kullanın ve her cihaza kendi kimlik bilgilerini verin. Client seçeneklerini daha derinlemesine anlamak isterseniz MQTTnet eğitimi TLS, MQTT 5 özellikleri ve yeniden bağlanma ayrıntılarını ele alır; MQTT istemci kütüphaneleri rehberi ise .NET'i diğer ekosistemlerle karşılaştırır. Broker'ı nerede barındıracağınızı değerlendiriyorsanız bulutta MQTT broker çalıştırma hakkındaki yazıyı okuyun.

ConnectMQ ile Başlayın

app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun, üretilen kimlik bilgilerini user secrets'a ekleyin; ASP.NET Core MQTT servisiniz ilk günden üretime uygun yapıda olsun. Backend'i geliştirirken cihaz trafiğini izlemek için portalın canlı topic gezginini kullanın.

Sık Sorulan Sorular

MQTT için IHostedService mi yoksa BackgroundService mi kullanmalıyım?

BackgroundService daha basit seçimdir, çünkü size durdurma token'ı olan tek bir ExecuteAsync metodu verir. IHostedService arayüzünü sizin yerinize uygular; bu nedenle ikisi de AddHostedService ile aynı şekilde kaydedilir.

DbContext neden doğrudan background service'e enjekte edilmemeli?

Hosted service'ler singleton, DbContext ise scoped'dur. Doğrudan enjekte etmek ya doğrulamada başarısız olur ya da tek bir context'i sonsuza kadar canlı tutar; bu yüzden IServiceScopeFactory ile her toplu işlem için bir scope oluşturun.

Her mesajı iki kez işlemeden birden fazla örneği nasıl çalıştırırım?

$share/devicehub/connectmq/+/telemetry gibi bir MQTT 5 shared subscription kullanın. Broker bu durumda her mesajı grubun yalnızca bir üyesine teslim eder.

Bir controller'dan cihazlara komut yayınlayabilir miyim?

Evet. Bağlı IMqttClient örneğini (veya küçük bir sarmalayıcıyı) singleton olarak kaydedin ve uç noktanızdan PublishAsync çağırın; her HTTP isteği için yeni bir bağlantı oluşturmayın.