C# ve MQTTnet 4/5 ile MQTT: Bağlanma, Abonelik, Yeniden Bağlanma

MQTTnet, .NET için standart MQTT kütüphanesidir. C# ile istemci yazmak için factory'den bir IMqttClient oluşturun, bağlantıyı MqttClientOptionsBuilder ile tanımlayın, bir ApplicationMessageReceivedAsync handler'ı ekleyin ve ardından ConnectAsync, SubscribeAsync ve PublishAsync çağırın. MQTTnet 5'ten itibaren managed client yok; yeniden bağlanmayı kendiniz yazarsınız, bu da yaklaşık yirmi satır tutar.

Bu rehber MQTTnet 5'i kullanıyor ve sürüm 4'ün farklı olduğu her noktayı belirtiyor; böylece iki sürümle de takip edebilirsiniz. Tüm örnekler broker.connectmq.com'a bağlanır ve testmqtt/dotnet/ altındaki topic'leri kullanır.

Kurulum ve sürüm seçimi

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

Bu komut en son major sürümü kurar. Mevcut bir MQTTnet 4 kod tabanını sürdürüyorsanız, geçişe hazır olana kadar --version 4.* ile sabitleyin. İlk karşılaşacağınız farklar:

ConvertPayloadToString() iki sürümde de çalışır; örneklerde bu yüzden kullanılıyor.

Bağlanma

using MQTTnet;            // v4: ayrıca `using MQTTnet.Client;` ekleyin
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($"Bağlandı: {result.ResultCode}");

Paylaşılan test endpoint'inde WithCredentials satırını atlayabilirsiniz; kendi broker'ınızda kullanıcı adı ve şifreyi koda gömmek yerine ortam değişkenlerinden verin. Broker bağlantıyı reddederse ConnectAsync, MqttClientConnectingFailedException fırlatır; bu exception'ın ResultCode değeri (örneğin BadUserNameOrPassword veya NotAuthorized) nedenini söyler. MQTT reason code referansı tüm değerleri listeler, en yaygın exception'lar ise MQTTnet hataları yazısında açıklanıyor.

Last Will ekleme

Çoğu istemci ortadan kaybolduğunu duyurmalıdır. Aynı options builder'a bir will mesajı ekleyin:

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

Broker will mesajını yalnızca bağlantı DISCONNECT paketi olmadan bittiğinde yayınlar: çökme, ağ kaybı veya keep-alive zaman aşımı. Düzgün bir DisconnectAsync() will'i bastırır; bu yüzden bağlandıktan sonra retained bir online durumunu, planlı kapatmadan önce de offline durumunu kendiniz yayınlayın. MQTT 5'te .WithWillDelayInterval(10), will gönderilmeden önce istemciye geri gelmesi için birkaç saniye tanır; bu da kısa ağ kesintilerinde durumun yanıp sönmesini önler. Ayrıntılar MQTT Last Will yazısında.

Abone olma ve mesajları işleme

Handler'ı abone olmadan önce kaydedin; aksi halde SUBACK'ten hemen sonra gelen retained mesajlar, henüz kimse dinlemezken ulaşabilir:

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

Subscribe sonucunu her zaman yazdırın. ACL nedeniyle bir aboneliği reddeden broker bağlantıyı kapatmaz; SUBACK içinde NotAuthorized gibi bir hata kodu döndürür ve handler'ınız hiç tetiklenmez. ResultCode'u kontrol etmek sessiz bir hatayı log satırına dönüştürür.

Handler, MQTTnet'in dahili dispatcher'ında çalışır. Kısa tutun: parse edin, işi devredin (bir Channel<T>'ye, kuyruğa veya arka plan task'ına) ve dönün. Handler içindeki uzun işler, acknowledgement'ları ve arkadan gelen tüm mesajları geciktirir.

Publish

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")      // yalnızca MQTT 5
    .WithUserProperty("source", "mqttnet-demo") // yalnızca MQTT 5
    .Build();

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

QoS 1'de PublishAsync, PUBACK geldiğinde tamamlanır; dolayısıyla reason code broker'ın cevabıdır. QoS 0'da ise paket sokete yazılır yazılmaz tamamlanır. Broker, maksimum paket boyutu nedeniyle büyük payload'ları reddedebilir; paket boyutu hesaplayıcı bir mesajın kablo üzerindeki gerçek boyutunu gösterir.

TLS ve WebSocket

Options builder'da taşıma katmanını değiştirin; geri kalan her şey aynı kalır:

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

Broker sertifikası işletim sisteminin güven deposuyla doğrulanır. Bir handshake hatasını yok etmek için WithCertificateValidationHandler(_ => true) eklemeye kapılmayın; nedeni bulun ve işe terminalden TLS bağlantısını test ederek başlayın. Eski MQTTnet 4 sürümleri artık obsolete olan .WithTls() metodunu kullanıyordu.

ManagedMqttClient olmadan yeniden bağlanma

MQTTnet 4, sizin yerinize yeniden bağlanan ve mesajları kuyruğa alan ManagedMqttClient'ı sunuyordu. Sürüm 5'te kaldırıldı ve resmi örnekler bunun yerine basit bir döngü kullanıyor. İşte eksiksiz bir sürüm:

client.ConnectedAsync += async e =>
{
    // Clean start ile broker aboneliklerimizi unuttu; kalıcı oturumda unutmamış olabilir.
    if (!e.ConnectResult.IsSessionPresent)
        await client.SubscribeAsync(subscribe, CancellationToken.None);
    Console.WriteLine("Bağlandı, oturum mevcut: " + e.ConnectResult.IsSessionPresent);
};

client.DisconnectedAsync += e =>
{
    Console.WriteLine($"Bağlantı kesildi ({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);           // başarılı bağlantıdan sonra backoff'u sıfırla
        }
    }
    catch (Exception ex) when (ex is not OperationCanceledException)
    {
        Console.WriteLine($"Yeniden bağlanma başarısız: {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();

Üç ayrıntı bunu sağlam kılar. Abonelik ConnectedAsync içinde durur; böylece yalnızca ilk bağlantıda değil, her yeniden bağlanmada geri yüklenir. Bekleme süresi biraz rastgele jitter ile üstel olarak artar; böylece broker'ı aynı anda kaybeden bin istemci aynı saniyede yeniden bağlanmaya çalışmaz. Bunun neden önemli olduğunu yeniden bağlanma en iyi uygulamaları anlatıyor. Ayrıca TryPingAsync bir health check görevi de görür: istemci bağlı değilken de, bağlantı yarı açık kaldığında da false döner.

Çevrimdışıyken tamponlama

Managed client giden mesajları da kuyruğa alıyordu. Buna ihtiyacınız varsa kodunuzla istemci arasına bir Channel<MqttApplicationMessage> koyun: üreticiler kanala yazar, tek bir tüketici task kanaldan okur ve yalnızca client.IsConnected true olduğunda PublishAsync çağırır; publish exception fırlatırsa mesajı yeniden dener. Uzun bir kesintinin belleği tüketmemesi için kanalı sınırlayın (Channel.CreateBounded) ve kanal dolduğunda en eski mi yoksa en yeni mesajların mı atılacağına bilinçli karar verin.

Kullanmaya değer MQTT 5 özellikleri

WithProtocolVersion(MqttProtocolVersion.V500) ayarlandığında bazı seçenekler hemen işe yarar:

• Session expiry — .WithSessionExpiryInterval(3600) ile .WithCleanStart(false) birlikte, bağlantı koptuktan sonra abonelikleri ve kuyruktaki QoS 1 mesajlarını bir saat saklar. Bunu özel bir broker'da test edin; paylaşılan test endpoint'leri genellikle çevrimdışı oturum saklamaz.
• Reason code ve reason string — connect, subscribe ve publish sonuçları ResultCode/ReasonCode ve çoğu zaman bir ReasonString içerir. Bunları loglayın; "çalışmıyor" ifadesini "kota aşıldı"ya çevirirler.
• User property — payload formatını değiştirmeden trace ID veya firmware sürümü gibi anahtar/değer metadata'sı.
• Message expiry — bir mesajda .WithMessageExpiryInterval(60), bir saat sonra yeniden bağlanan cihaza eskimiş komutların teslim edilmesini engeller.
• Request/response — .WithResponseTopic() ve .WithCorrelationData(), MQTT üzerinden RPC uygulamanın standart yolunu sunar.

Test edin

Konsol uygulamasını çalıştırın, online MQTT istemcisini açın ve testmqtt/dotnet/# topic'ine abone olun. JSON ölçüm değerinin testmqtt/dotnet/sensors/lab üzerinde geldiğini görmelisiniz. Tarayıcıdan testmqtt/dotnet/sensors/test topic'ine publish yapın; C# handler'ı mesajı yazdırır. Reconnect döngüsünü test etmek için ağ bağlantınızı birkaç saniye kesin ve log'un Bağlantı kesildi satırından Bağlandı, oturum mevcut: False satırına ve ardından yeni bir SUBACK'e geçtiğini izleyin.

MQTTnet istemcinizin mesajlarını canlı izleyin

Bir web API veya worker service geliştiriyorsanız sonraki adım, bu istemciyi dependency injection ve health check'lerle birlikte bir hosted service'e taşımaktır: ASP.NET Core'da MQTT yazısına bakın.

MQTTnet size küçük ve açık bir API sunar: bir options builder, üç async çağrı ve birkaç event. ConnectedAsync içinde yeniden abone olan bir reconnect döngüsü ekleyin, her sonuç kodunu loglayın; aynı istemci paylaşılan endpoint'teki bir konsol testinden kendi broker'ınızdaki production servisine kadar çalışır.

Sıkça sorulan sorular