簡(jiǎn)介MQTT客戶(hù)端C#版是一份面向物聯(lián)網(wǎng)開(kāi)發(fā)者與C#學(xué)習(xí)者的實(shí)戰(zhàn)項(xiàng)目源碼基于M2Mqtt.Net.dll實(shí)現(xiàn)MQTT協(xié)議通信可用于智能家居、工業(yè)自動(dòng)化、遠(yuǎn)程監(jiān)測(cè)等場(chǎng)景的上位機(jī)開(kāi)發(fā)。資源包共35個(gè)文件約223KB以cs源碼、dll類(lèi)庫(kù)、resx資源、exe可執(zhí)行文件及config配置為主另含sln解決方案與csproj工程文件結(jié)構(gòu)完整便于直接編譯運(yùn)行與二次開(kāi)發(fā)。項(xiàng)目涵蓋連接管理、主題訂閱、消息發(fā)布、心跳?；钆c異常重連等核心邏輯并配有WinForm界面設(shè)計(jì)文件可幫助讀者理解C#如何與MQTT庫(kù)集成、構(gòu)建圖形化上位機(jī)及處理MQTT事件。目前已有4171人學(xué)習(xí)下載適合希望掌握MQTT協(xié)議原理與C#客戶(hù)端實(shí)現(xiàn)方法的開(kāi)發(fā)者參考借鑒。1. 從一次設(shè)備離線(xiàn)告警說(shuō)起MQTT 客戶(hù)端 C# 版到底能干什么產(chǎn)線(xiàn)上十幾臺(tái)工控機(jī)跑著數(shù)據(jù)采集上位機(jī)需要實(shí)時(shí)拿到每臺(tái)設(shè)備的溫度、轉(zhuǎn)速和報(bào)警狀態(tài)。最初用 HTTP 輪詢(xún)2 秒一次設(shè)備一多服務(wù)端就喘網(wǎng)絡(luò)抖一下還容易丟狀態(tài)。后來(lái)?yè)Q成 MQTT發(fā)布訂閱模型一上帶寬和實(shí)時(shí)性都順了但新的問(wèn)題來(lái)了——服務(wù)端是 C# 寫(xiě)的得有一個(gè)能長(zhǎng)期穩(wěn)定跑在 Windows 服務(wù)里的 MQTT 客戶(hù)端。這就是這份「MQTT 客戶(hù)端 C# 版」資源要解決的事它不是一個(gè)玩具 Demo而是一個(gè)能訂閱、發(fā)布、斷線(xiàn)重連、處理 QoS 的客戶(hù)端實(shí)現(xiàn)適合做工業(yè)上位機(jī)、物聯(lián)網(wǎng)網(wǎng)關(guān)、數(shù)據(jù)采集后臺(tái)的 .NET 開(kāi)發(fā)者。MQTT 本身是輕量級(jí)的消息傳輸協(xié)議基于 TCP走發(fā)布/訂閱模式核心概念就幾個(gè)Broker消息代理、Topic主題、QoS服務(wù)質(zhì)量等級(jí)、Client ID客戶(hù)端標(biāo)識(shí)。C# 這邊主流用 MQTTnet 這個(gè)庫(kù)跨平臺(tái)、異步、支持 MQTT 3.1.1 和 5.0。這份資源的價(jià)值在于把「連上、訂閱、收消息、發(fā)消息、斷線(xiàn)重連」這條鏈路用可運(yùn)行的代碼串起來(lái)而不是讓你對(duì)著官方文檔一行行猜。下面從環(huán)境搭建講到參數(shù)配置再到實(shí)際會(huì)翻車(chē)的地方照著走能少踩不少坑。2. 環(huán)境準(zhǔn)備與 MQTTnet 選型為什么不是別的庫(kù)2.1 三個(gè) C# MQTT 庫(kù)的取舍C# 生態(tài)里能用的 MQTT 客戶(hù)端庫(kù)不多常見(jiàn)的有 MQTTnet、M2MqttGnatMQ 系、以及一些封裝了 Paho 的綁定。選型時(shí)我一般看四點(diǎn)是否支持 MQTT 5.0、異步 API 是否干凈、斷線(xiàn)重連是否內(nèi)置、社區(qū)是否還在維護(hù)。庫(kù)MQTT 5.0異步支持?jǐn)嗑€(xiàn)重連維護(hù)狀態(tài)適用場(chǎng)景MQTTnet支持原生 async/await需自己寫(xiě)活躍新項(xiàng)目首選M2Mqtt不支持回調(diào)為主部分基本停更老項(xiàng)目維護(hù)Paho 綁定部分一般需封裝一般跨語(yǔ)言統(tǒng)一結(jié)論很直接新項(xiàng)目用 MQTTnet。它的 API 設(shè)計(jì)貼近 .NET 習(xí)慣MqttFactory創(chuàng)建客戶(hù)端ConnectAsync、SubscribeAsync、PublishAsync都是異步的配合CancellationToken能干凈地退出。M2Mqtt 那套事件回調(diào)在 .NET Core 里用起來(lái)別扭而且不支持 5.0 的新特性。2.2 建項(xiàng)目、裝包、確認(rèn)版本先建一個(gè) .NET 控制臺(tái)或 Worker Service 項(xiàng)目。工業(yè)場(chǎng)景我傾向 Worker Service因?yàn)樗焐m合跑后臺(tái)長(zhǎng)駐任務(wù)。# 建一個(gè) Worker Service 項(xiàng)目適合做后臺(tái)常駐客戶(hù)端 dotnet new worker -n MqttClientDemo cd MqttClientDemo # 安裝 MQTTnet注意版本4.x 和 3.x 的 API 差異較大 dotnet add package MQTTnet --version 4.3.7.1207裝完確認(rèn)一下csproj里的引用別裝成 3.x 了3.x 的MqttClientOptionsBuilder部分方法簽名和 4.x 不一樣照著 4.x 的代碼寫(xiě)會(huì)編譯不過(guò)。這是第一個(gè)容易翻車(chē)的地方。提示MQTTnet 4.x 把很多擴(kuò)展方法挪到了MQTTnet.Extensions命名空間下WithTcpServer、WithCredentials這些如果找不到先檢查 using 是否齊全。2.3 Broker 從哪來(lái)客戶(hù)端要連一個(gè) Broker。測(cè)試階段常見(jiàn)做法是本地起一個(gè)用 EMQX 或 Mosquitto 都行Docker 一條命令就能跑起來(lái)# 本地起一個(gè) Mosquitto映射 1883 端口適合開(kāi)發(fā)調(diào)試 docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto:2生產(chǎn)環(huán)境 Broker 通常是獨(dú)立部署的客戶(hù)端只需要拿到地址、端口、用戶(hù)名密碼。這里要注意Broker 的allow_anonymous配置如果開(kāi)著本地測(cè)試能連上一上生產(chǎn)關(guān)了匿名就連接失敗別到那時(shí)候才查。3. 連接、訂閱、發(fā)布把核心鏈路跑通3.1 構(gòu)建客戶(hù)端與連接參數(shù)連接是整條鏈路的地基參數(shù)設(shè)錯(cuò)后面全白搭。下面這段是連接的核心代碼我把它拆開(kāi)講。using MQTTnet; using MQTTnet.Client; // 用工廠(chǎng)創(chuàng)建客戶(hù)端實(shí)例一個(gè)客戶(hù)端對(duì)應(yīng)一個(gè)連接 var factory new MqttFactory(); var client factory.CreateMqttClient(); // 構(gòu)建連接選項(xiàng)這里每一項(xiàng)都影響連接行為 var options new MqttClientOptionsBuilder() .WithTcpServer(127.0.0.1, 1883) // Broker 地址和端口 .WithClientId(gateway-001) // 客戶(hù)端唯一標(biāo)識(shí)重復(fù)會(huì)頂?shù)羟耙粋€(gè) .WithCredentials(user, pass) // 用戶(hù)名密碼匿名 Broker 可省略 .WithCleanSession(true) // 是否清除會(huì)話(huà)見(jiàn)下方說(shuō)明 .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) // 心跳間隔 .Build(); // 連接失敗會(huì)拋異常生產(chǎn)環(huán)境要 try-catch var result await client.ConnectAsync(options, CancellationToken.None); Console.WriteLine($連接結(jié)果: {result.ResultCode});WithClientId是重點(diǎn)。同一個(gè) Client ID 同時(shí)連兩次Broker 會(huì)把前一個(gè)踢掉現(xiàn)象就是「客戶(hù)端莫名其妙掉線(xiàn)」。工業(yè)場(chǎng)景里每臺(tái)設(shè)備用固定且唯一的 ID比如用設(shè)備序列號(hào)拼出來(lái)。WithCleanSession(true)表示每次連接都是干凈會(huì)話(huà)Broker 不保留訂閱關(guān)系和離線(xiàn)消息如果要收離線(xiàn)消息得設(shè)成false并且用固定的 Client ID否則會(huì)話(huà)對(duì)不上。KeepAlivePeriod是心跳設(shè)太短網(wǎng)絡(luò)開(kāi)銷(xiāo)大設(shè)太長(zhǎng) Broker 可能判定你掉線(xiàn)30 秒是個(gè)穩(wěn)妥值。3.2 訂閱主題與通配符訂閱要理解 Topic 層級(jí)和通配符。匹配單層#匹配多層。比如factory/line1//temp能匹配factory/line1/machineA/temp但匹配不了factory/line1/machineA/room1/temp。// 訂閱一個(gè)帶通配符的主題QoS 用 AtLeastOnce var subOptions new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(f f .WithTopic(factory/line1//status) // 匹配單層 .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)) .Build(); await client.SubscribeAsync(subOptions, CancellationToken.None);訂閱完要掛消息接收事件否則消息來(lái)了你收不到// 消息到達(dá)時(shí)觸發(fā)e.ApplicationMessage 里是負(fù)載 client.ApplicationMessageReceivedAsync e { var topic e.ApplicationMessage.Topic; var payload System.Text.Encoding.UTF8.GetString( e.ApplicationMessage.PayloadSegment); // 4.x 用 PayloadSegment Console.WriteLine($收到 [{topic}]: {payload}); return Task.CompletedTask; };注意 4.x 里取負(fù)載用的是PayloadSegment不是老版本的Payload這個(gè)改動(dòng)坑過(guò)不少人編譯報(bào)錯(cuò)時(shí)先看這里。3.3 發(fā)布消息與 QoS 選擇發(fā)布消息時(shí) QoS 決定可靠性。QoS 0 最多一次可能丟QoS 1 至少一次可能重復(fù)QoS 2 恰好一次開(kāi)銷(xiāo)最大。工業(yè)數(shù)據(jù)采集里狀態(tài)上報(bào)用 QoS 1 比較平衡控制指令如果要求不重復(fù)可以用 QoS 2但別濫用QoS 2 的握手流程會(huì)拖慢吞吐。// 構(gòu)建并發(fā)布一條消息 var message new MqttApplicationMessageBuilder() .WithTopic(factory/line1/machineA/status) .WithPayload({\temp\:36.5,\rpm\:1200}) // 負(fù)載建議用 JSON .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce) .WithRetainFlag(false) // 是否保留最后一條消息 .Build(); await client.PublishAsync(message, CancellationToken.None);WithRetainFlag值得單獨(dú)說(shuō)。設(shè)成 trueBroker 會(huì)保留這條消息新訂閱者一連上立刻收到最后一條。適合「設(shè)備當(dāng)前狀態(tài)」這種場(chǎng)景不適合「一次報(bào)警事件」否則新訂閱者會(huì)收到一條早就過(guò)期的報(bào)警。3.4 斷線(xiàn)重連別指望庫(kù)幫你全搞定MQTTnet 不會(huì)自動(dòng)重連得自己監(jiān)聽(tīng)DisconnectedAsync事件然后重連。這是生產(chǎn)環(huán)境必須寫(xiě)的部分。// 斷線(xiàn)事件里做重連加退避避免瘋狂重試 client.DisconnectedAsync async e { Console.WriteLine($斷開(kāi): {e.Reason}); // 簡(jiǎn)單退避實(shí)際項(xiàng)目建議指數(shù)退避 await Task.Delay(TimeSpan.FromSeconds(5)); try { await client.ConnectAsync(options, CancellationToken.None); Console.WriteLine(重連成功); } catch (Exception ex) { Console.WriteLine($重連失敗: {ex.Message}); } };這段代碼有個(gè)隱患如果重連也失敗不會(huì)再次觸發(fā)重連客戶(hù)端就永久掉線(xiàn)了。穩(wěn)妥做法是套一個(gè)循環(huán)加指數(shù)退避或者用while直到連上為止。這是新手最容易忽略的地方測(cè)試時(shí)拔網(wǎng)線(xiàn)再插上往往就發(fā)現(xiàn)客戶(hù)端再也連不回來(lái)了。4. 避坑與排查那些讓你加班到半夜的問(wèn)題4.1 現(xiàn)象客戶(hù)端頻繁掉線(xiàn)日志里全是重連原因通常有三個(gè)。一是 Client ID 重復(fù)兩個(gè)進(jìn)程用了同一個(gè) ID互相頂。二是 KeepAlive 設(shè)得太短網(wǎng)絡(luò)稍有延遲 Broker 就判定超時(shí)。三是 Broker 端有max_keepalive限制客戶(hù)端設(shè)的值超過(guò)了它。排查時(shí)先看 Broker 日志它會(huì)記錄踢掉客戶(hù)端的原因再確認(rèn) Client ID 是否唯一最后把 KeepAlive 調(diào)到 60 秒試試。4.2 現(xiàn)象訂閱了但收不到消息先確認(rèn) Topic 拼寫(xiě)和通配符層級(jí)對(duì)不對(duì)factory//status和factory/line1/status是兩回事。再確認(rèn)發(fā)布方用的 Topic 是否真的匹配。還有一個(gè)隱蔽原因訂閱時(shí) QoS 和發(fā)布時(shí) QoS 不一致某些 Broker 在特定配置下會(huì)降級(jí)投遞。排查手段是用一個(gè)通用的#訂閱所有主題看消息到底有沒(méi)有到 Broker能到就是訂閱過(guò)濾的問(wèn)題不能到就是發(fā)布端的問(wèn)題。4.3 現(xiàn)象消息重復(fù)收到QoS 1 的語(yǔ)義就是「至少一次」重復(fù)是正常的不是 bug。解決方式是在應(yīng)用層做冪等比如消息里帶一個(gè)唯一 ID收到后去重。別指望把 QoS 改成 2 就萬(wàn)事大吉QoS 2 開(kāi)銷(xiāo)大而且實(shí)現(xiàn)不當(dāng)一樣可能出問(wèn)題。工業(yè)場(chǎng)景里我一般用 QoS 1 加業(yè)務(wù)層去重比 QoS 2 更可控。4.4 現(xiàn)象程序退出時(shí)連接沒(méi)斷干凈client.DisconnectAsync()沒(méi)調(diào)或者調(diào)了沒(méi) await。進(jìn)程退出時(shí) TCP 連接可能還掛著B(niǎo)roker 要等 KeepAlive 超時(shí)才清理。正確做法是在StopAsync或退出鉤子里 await 斷開(kāi)并且給一個(gè)超時(shí)別讓斷開(kāi)卡住整個(gè)退出流程。// 優(yōu)雅斷開(kāi)帶超時(shí)保護(hù) using var cts new CancellationTokenSource(TimeSpan.FromSeconds(3)); await client.DisconnectAsync(new MqttClientDisconnectOptions(), cts.Token);4.5 現(xiàn)象大負(fù)載消息導(dǎo)致內(nèi)存飆升MQTT 單條消息默認(rèn)有大小限制Broker 和客戶(hù)端都有。發(fā)大 JSON 或二進(jìn)制時(shí)如果超過(guò)限制會(huì)被斷開(kāi)。排查時(shí)看 Broker 的message_size_limit配置客戶(hù)端這邊也要注意別把整個(gè)文件塞進(jìn)一條消息。常見(jiàn)做法是大數(shù)據(jù)分片或者只發(fā)一個(gè)引用地址讓接收方自己去拉。5. 進(jìn)階把客戶(hù)端做成能長(zhǎng)期跑的服務(wù)5.1 用 Worker Service 托管生命周期控制臺(tái)程序一關(guān)就沒(méi)了生產(chǎn)環(huán)境得用 Worker Service 或 Windows 服務(wù)。把 MQTT 客戶(hù)端放進(jìn)BackgroundService的ExecuteAsync里配合IHostApplicationLifetime處理退出。public class MqttWorker : BackgroundService { private readonly IMqttClient _client; private readonly MqttClientOptions _options; public MqttWorker(IMqttClient client, MqttClientOptions options) { _client client; _options options; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { // 掛事件、連接、訂閱都在這里 _client.ApplicationMessageReceivedAsync OnMessageAsync; await _client.ConnectAsync(_options, stoppingToken); // 阻塞直到服務(wù)停止 await Task.Delay(Timeout.Infinite, stoppingToken); } private Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs e) { // 處理消息注意別在這里做耗時(shí)操作會(huì)阻塞接收 return Task.CompletedTask; } }關(guān)鍵點(diǎn)OnMessageAsync里別做耗時(shí)操作MQTTnet 的消息處理是串行的你在這里卡住后面的消息全堵著。要處理耗時(shí)邏輯就丟到隊(duì)列里用另一個(gè)線(xiàn)程消費(fèi)。5.2 消息處理與背壓高吞吐場(chǎng)景下消息來(lái)得比處理快內(nèi)存會(huì)漲。常見(jiàn)做法是引入Channel做緩沖接收事件只負(fù)責(zé)往 Channel 里寫(xiě)后臺(tái)任務(wù)慢慢消費(fèi)。// 有界 Channel滿(mǎn)了就丟或阻塞防止內(nèi)存無(wú)限增長(zhǎng) private readonly ChannelMqttApplicationMessage _channel Channel.CreateBoundedMqttApplicationMessage(1000); // 接收事件里只寫(xiě) Channel不做業(yè)務(wù) private async Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs e) { await _channel.Writer.WriteAsync(e.ApplicationMessage); }CreateBounded的容量根據(jù)業(yè)務(wù)定1000 是個(gè)起點(diǎn)。滿(mǎn)了之后WriteAsync會(huì)等待等于給上游一個(gè)背壓信號(hào)。如果不想等可以用TryWrite滿(mǎn)了就丟但要記錄丟棄數(shù)量否則數(shù)據(jù)丟了都不知道。5.3 驗(yàn)證客戶(hù)端是否真的穩(wěn)寫(xiě)完別急著上線(xiàn)做幾個(gè)驗(yàn)證。拔網(wǎng)線(xiàn) 30 秒再插上看是否自動(dòng)重連并恢復(fù)訂閱。用mosquitto_pub手動(dòng)發(fā)消息確認(rèn)能收到。把 Broker 重啟看客戶(hù)端行為。壓測(cè)時(shí)用腳本每秒發(fā)幾百條觀(guān)察內(nèi)存和 CPU。我一般還會(huì)在客戶(hù)端里加一個(gè)心跳主題定期發(fā)布自己的存活狀態(tài)這樣監(jiān)控端能一眼看出哪個(gè)客戶(hù)端掉線(xiàn)了。從那以后我每次寫(xiě) MQTT 客戶(hù)端都會(huì)先把斷線(xiàn)重連和優(yōu)雅退出這兩段代碼寫(xiě)死再動(dòng)業(yè)務(wù)邏輯——這兩個(gè)地方翻車(chē)后面查起來(lái)就是黑匣子。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取