Two overnight failures reveal the core of MQTT reliability: reconnection requires resubscription, and retrying is necessary to prevent panic. The root cause lies in ignoring the diversity of disconnected connections, especially “half-open connections” that lead to false online connections; client IDs should be unique and stable to avoid session loss or mutual kicking; heartbeats (KeepAlive) need to be configured in layers to avoid misjudgment and waste of power consumption. Only by combining client generation specifications and Broker side management and control, supplemented by application layer heartbeats, can we build a truly reliable IoT communication link.
1.1. The first all-nighter: collective holiday online.
An industrial project, on-site optical fiber cutover in the middle of the night. Monitor 380 devices on the big screenturn gray at the same time——But what’s even weirder is that a few hours later they all said “I’m back” again. The first reaction of the colleagues on duty was “the network has been restored”, but this is not the case: the equipment is actuallyIt’s been reconnected a long time agojust after connectingNo resubscription to topicso the backend could not receive a single message; and the online status of the large screen relied on the device to report it, so the whole line was in ashes and would not be restored until the next restart process.
Root cause: The reconnection code only writes ConnectAsync().Didn’t write resubscription. The connection was connected, but all subscription relationships were lost.
1.2. The second all-nighter: Reconnection Storm.
Another project, 2000 devices. One time the Broker jittered, and the clients of all devices detected disconnection at the same time, and then usedCrazy retries at the same fixed interval (5 seconds). Result: As soon as the Broker recovered, it was hung up again by connection requests from 2,000 devices in the same second, and so on – classicthundering herd。
These two things point to the same conclusion:The connection reliability of MQTT is not “whether it can be reconnected”, but “whether it is resubscribed when reconnecting, and whether retrying will interrupt the service.”
Both of these points must be considered clearly during the design stage.
Before writing any code, figure out who your enemy is. Broken connections don’t just look like one.
2.1 Four types of disconnection
It also shows up as “connection unavailable”, but the triggering reasons, protocol characteristics and client perceptibility vary greatly – especially be wary of “half-open connections”.
2.2 The most insidious one: half-open connection
The essence of a half-open connection is:Both ends of the TCP connection have inconsistent knowledge of the state. The client thinks that the connection is still there, and the Broker has released the session; or conversely, the Broker thinks that the client is still online, but the intermediate link is no longer available.
In this case:
-
IsConnected returns true – because the TCP layer did not receive a disconnect signal
-
No error is reported when sending a message – the data is written to the local sending buffer, but it never reaches the peer.
-
DisconnectedAsync Not triggered——Because there is no disconnect event to capture
The device is “online” but does nothing. It lasts for more than ten hours. This is where the fake line in the introduction comes from.
2.3 Three-sentence conclusion
-
DisconnectedAsync trigger ≠ disconnection must be sensed(It doesn’t trigger at all with a half-open connection)
-
IsConnected == true ≠ The connection must be alive(Always returns true when false online)
-
ConnectAsync successful ≠ You can subscribe immediately(In the reconnection scenario, the internal state may not be ready yet)
Based on these three cognitions, let’s break them down one by one.
This is the easiest one to ignore, but the easiest one to bury. Many people only understand ClientId by saying “it must be unique”, so they write it like this:
// ❌ 反面示例1：带时间戳，每次重连都是"新设备"
var clientId = $"device-{deviceId}-{DateTime.Now.Ticks}";
// ❌ 反面示例2：每次启动随机生成
var clientId = $"device-{Guid.NewGuid()}";
3.1. Pitfall 1: Every time you reconnect, it will be a “new client”, and the session will never be renewed.
In MQTT, Broker relies on ClientId to identify “whether this is the same session”. You bring a timestamp or a random number,Every time you reconnect, it is a new device in the eyes of the Broker.: The old session becomes an unclaimed zombie session, and the new session starts from scratch. The network trembled, hundreds of zombie sessions piled up, and the upper limit of the number of connections was directly exceeded.
3.2. Pitfall 2: Multiple devices share the same ClientId = kick each other.
Conversely, if two devices use the same ClientId (for example, the factory configuration script bug has device-001 written in both), the Broker’s rules are:New connections will kick out old connections. So the two devices kicked each other off endlessly, and no one was able to connect stably.
Real case: In a smart park project, more than 300 access control devices were all displayed offline. The investigation was carried out until 6 am. The reason was that the two devices had burned the same ClientId. The one connected later kicked out the one connected first, and the one connected first reconnected and then kicked the one connected later——infinite loop. The Broker log only shows “client disconnected by new connection” without any warning.
3.3. Three iron rules
ClientId is not a random string – it determines session continuation, kicking behavior and later maintainability.
3.4. Recommended naming paradigm
Naming is not “as long as it runs”, but a prerequisite for connection reliability, ACL control, and troubleshooting. A good name allows you to locate the device identity at a glance in the Broker management background, instead of being confused by a bunch of device-xxxx.
格式：{产品码}_{环境}_{位置}_{序列号}
示例：
gw_prod_parkA_001 # 园区A的1号网关
sn_SH202610190001 # 上海2026年10月19日第1台设备
hvac_beta_floor3_001 # 测试环境3楼空调
Constraint suggestions:
-
The length is controlled at Within 23 characters. In the MQTT 3.1.1 era, Broker had a 23-character limit on ClientId length. Although it was relaxed in MQTT 5.0, it is still recommended to be conservative for cross-version and cross-Broker compatibility.
-
The serial number is written into the local configuration when the device leaves the factory or is first activated.Never change after that。
-
Environment fields (prod/staging/dev) allow the same set of ACL rules to be reused across environments.
3.5. Absolutely prohibited:
These traps me in the production environmentstepped on allbehind each one is a late-night warning. Remember: Once the ClientId is written incorrectly, the problem will not be exposed immediately – it will explode in the weirdest way when you are most relaxed.
3.6. Dynamic generation + Broker side management and control
The client’s own generated ClientId is only the first line of defense, Broker You must know everything. No matter how good the naming convention is, it cannot stop bugs in the factory burning script, random strings typed by testers, or “temporary changes” by a new colleague one day. The management and control strategy on the Broker side is the last gate – blocking all non-compliant connections.
// 设备端：用 MAC + 产品型号生成，烧录阶段固化
public static string GenerateClientId(string productCode, string macAddress)
{
var macSuffix = macAddress.Replace(":", "").Replace("-", "").TakeLast(6);
return $"{productCode}-{new string(macSuffix.ToArray())}";
}
// 示例输出：hvac-A1B2C3
# EMQX 服务端管控（防止意外重复）
# 1. 启用 ClientId 白名单
# 2. 设置最大连接数 per ClientId = 1
# 3. 设置 ClientId 正则校验：^[a-z0-9\-]{3,23}$
One sentence summary: Client generation is guaranteed to be “right”, Broker management and control is guaranteed to be “correct” – when the two layers are added together, the ClientId line is truly welded.
4.1 Working principle
KeepAlive is a number of seconds declared by the client in the CONNECT message. After that, as long as within this period of timeNo other messages were sentthe client must send a PINGREQ; the Broker returns PINGRESP. in turn,If the Broker does not receive any message from the client within 1.5 × KeepAlive time, it will be judged to be offline.。
Business messages will refresh the heartbeat timing – in high-frequency reporting scenarios, the business data itself serves as the heartbeat, and PINGREQ is almost never sent.
4.2 Hierarchical configuration table (by network + power supply)
There is no “universal value” to KeepAlive. If the setting is too short, PINGRESP will not be able to return in time under a weak network, and it will be misjudged to be offline, which will increase the burden of reconnection; if the setting is too long, it will take a long time to find out after the actual disconnection, and the false online time will be extended. The correct approach is to layer configurations by network type and power supply method – the core trade-off lies in two variables: latency fluctuations and power consumption budget.
Value based on network type and power consumption constraints: Broker side timeout is usually 1.5 times KeepAlive, leaving a margin for one packet loss.
4.3 Three iron rules
There are only three pitfalls in the heartbeat parameters, but each of them can make the system “seem to be usable, but in fact it will explode at any time.” These three are the bottom-line rules I summarized after countless online failures——Not “best practices”, but “something will happen if you don’t do this”。
-
KeepAlive ≠ 0： 0 means turning off the heartbeat. If there is no message for a long time, the NAT gateway directly cuts off TCP, and the Broker will never detect that the device is offline.
-
KeepAlive is not too small: The 5-second heartbeat PINGRESP cannot return in time under a weak network, and is mistakenly judged as offline, which in turn consumes more power.
-
When reporting high-frequency business, KeepAlive can be boldly set to a larger size: Anyway, business packets will refresh the heartbeat timing.
4.4 Configuration in MQTTnet
I talked about the principle and layering strategy of KeepAlive before, but here we come to the code level. MQTTnet’s MqttClientOptionsBuilder is the only entrance to connection parameters – the configuration of all three pillars is concentrated here. If one line is written incorrectly, everything will be lost later.
var options = new MqttClientOptionsBuilder()
.WithTcpServer("broker.example.com", 1883)
.WithClientId($"gw-prod-parkA-001")
.WithKeepAlivePeriod(TimeSpan.FromSeconds(30))
.WithCleanSession(false) // 保留会话，离线消息不丢
.Build();
4.5 ⚠️ Known defects + Application layer heartbeat protection
How many MQTTnets are there?Known issuesWill cause the KeepAlive mechanism to fail:
-
issue #2248： TrySendKeepAliveMessages returns directly after catching OperationCanceledException, DisconnectedAsync never triggersIsConnected always returns true. There is an extreme case in the official issue: the upstream Broker cluster did not send a TCP disconnect signal when replacing nodes, and the client continued to be in the “false online” state. 17 hours。
-
issue #2180： After the reconnection is successful, IsConnected is still false, causing the code to fall into a loop and reconnect.
-
issue #2244： DisconnectedAsync does not fire when the connection is interrupted.
Conclusion: KeepAlive cannot be trusted alone. The production environment must be protected by application layer heartbeats.
#region 应用层心跳
// 心跳循环：周期发送 MQTT PINGREQ；失败时主动断开连接，让重连流程接管。
private async Task HeartbeatLoopAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
// 1) 按配置间隔等待
await Task.Delay(_heartbeatInterval, ct);
// 2) 不在线则跳过（重连循环会负责恢复）
if (!_client.IsConnected)
{
continue;
}
// 3) 发送应用层心跳（PINGREQ），带超时避免半开连接永久挂起
bool pong;
try
{
using var pingCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
pingCts.CancelAfter(TimeSpan.FromSeconds(PingTimeoutSec));
pong = await _client.TryPingAsync(pingCts.Token);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
// 超时（非外部取消）：视为心跳失败
pong = false;
}
// 4) 心跳失败：连接可能已半开，强制断开以触发重连
if (!pong)
{
_logger.LogWarning("心跳探测失败，强制重连");
_ = ForceReconnectAsync();
}
}
}
// 主动断开当前连接，触发 DisconnectedAsync 事件进入重连流程。
private async Task ForceReconnectAsync()
{
try
{
// 仅在仍在线时才需要断开
if (_client.IsConnected)
{
await _client.DisconnectAsync();
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "强制断开连接失败");
}
}
#endregion
4.6 Session strategy: CleanSession vs CleanStart + SessionExpiry
If the connection is disconnected and then reconnected, do you still need the messages that were not received/confirmed before? This question determines your session strategy. If you choose the wrong one, you may lose your order at best, or you may miss out on billing at worst. MQTT 3.1.1 only has one switch: CleanSession, which is either black or white; MQTT 5.0 is split into two dimensions: CleanStart + SessionExpiryInterval, which brings the flexibility to a higher level – but it is also easier to mismatch.
The scenario determines whether the session is reused: telemetry refreshes the session without thinking, commands short-term continuation, and billing is permanently retained – but permanent sessions consume the most Broker resources.
pit: WithCleanSession() in MQTT 5.0 has no parameters = CleanStart=false (reuse old session); WithCleanSession(true) = CleanStart=true. Be sure to check the source code to confirm, don’t rely on guessing.
5.1. Why fixed interval = self-directed DDoS
If all devices take a fixed 5 seconds to reconnect, when the Broker briefly fluctuates, thousands of devices will rush up at the same time and hang up the newly restored service again. That’s how the second all-nighter came about.
5.2. Comparison of retreat strategies
The reconnection strategy essentially answers one question: “It’s disconnected, how long should I wait and try again?“ This answer determines whether your system will recover gracefully when the network jitters, or make things worse. The four strategies range from “streaking” to “production level”, and the gap is not even a tiny bit.
The core contradiction of the four strategies is to recover as quickly as possible without crushing the recovering Broker again.
5.3. Backoff parameter table
The previous section talked about “Why use exponential backoff + jitter”, this section focuses on specific numbers. Parameters are not determined by patting your head——Starting too aggressively will crash the Broker, capping too short will waste battery, capping too long will make the device “unable to come back”. The following set of parameters are recommended values after repeated tuning by me on a scale of 500 to 2000 devices. You can copy them directly.
The backoff limit is capped for 30 seconds, and 10%~30% random jitter is superimposed to avoid simultaneous retries of a large number of devices.
Note: The jitter coefficient is [1.1, 1.3] A uniform random number within the interval; the upper limit is fixed at 30s from the 6th time and will not be doubled to prevent the reconnection interval from increasing indefinitely.
Capping is a must: There is no upper limit. The device will have to wait dozens of minutes to come back after being disconnected for an hour, which is unacceptable.
5.4. Key pitfall: Reconnection successful ≠ now available for subscription
ConnectAsync returns successfully, which does not mean that all internal states are ready. Especially in reconnection scenarios, calling SubscribeAsync immediately may report “client not connected”, or the subscription may be silently ignored.
Correct approach: The client maintains a subscription registration table by itself and restores it one by one after reconnecting. Don’t rely on the Broker to restore everything for you.
// 连接成功回调：重置退避计数、恢复订阅、补发离线缓冲消息。
private async Task OnConnectedAsync(MqttClientConnectedEventArgs e)
{
// 1) 重置连续失败计数，下次重连从 base 开始
Interlocked.Exchange(ref _attempt, 0);
_logger.LogInformation("MQTT 连接成功");
// 2) 按订阅表逐个恢复订阅
foreach (var (topic, qos) in _subscriptions)
{
await _client.SubscribeAsync(
new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter(topic, qos)
.Build());
}
// 3) 补发断线期间缓冲的消息
while (_outbox.TryDequeue(out var msg))
{
try
{
await _client.PublishAsync(msg);
}
catch (Exception ex)
{
// 补发失败：重新入队并中断本次补发，等下次连接成功再试
_logger.LogWarning(ex, "补发离线消息失败，重新入队");
_outbox.Enqueue(msg);
break;
}
}
}
5.5. Another pitfall: reconnection loop superposition
The DisconnectedAsync event may fire multiple times. If there is no lock, there will be several reconnection cycles running at the same time.
// 重连锁：保证同一时刻只有一个重连循环在跑，避免并发重连。
private readonly SemaphoreSlim _reconnectLock = new(1, 1);
// 断线回调：由曾经在线触发的断线才启动重连循环，避免初始连接失败被重复触发。
private Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs e)
{
_logger.LogWarning("MQTT 连接断开：{Reason}", e.Reason);
// 仅当之前确实在线时才触发重连，避免初始连接失败被事件重复触发
if (e.ClientWasConnected)
{
_ = ReconnectLoopAsync();
}
return Task.CompletedTask;
}
// 重连循环：通过信号量保证同一时刻只有一个循环；按指数退避 + 随机抖动等待后重试。
private async Task ReconnectLoopAsync()
{
// 1) 非阻塞抢锁：已有人在重连则直接退出
if (!await _reconnectLock.WaitAsync(0))
{
return;
}
try
{
// 2) 循环重试，直到取消或连接成功
while (!_cts.IsCancellationRequested && !_client.IsConnected)
{
// 2.1 计算本次等待时长（指数退避 + 抖动）
var delay = NextDelay();
_logger.LogInformation(
"第 {Attempt} 次重连尝试，等待 {Delay:0.00} 秒",
_attempt,
delay.TotalSeconds);
// 2.2 退避等待，期间被取消则退出
try
{
await Task.Delay(delay, _cts.Token);
}
catch (OperationCanceledException)
{
return;
}
// 2.3 带超时地尝试连接
try
{
using var timeout = new CancellationTokenSource(
TimeSpan.FromSeconds(ReconnectTimeoutSec));
await _client.ConnectAsync(_options, timeout.Token);
// 连接成功，OnConnectedAsync 会重置 _attempt 并做重订阅/补发
return;
}
catch (Exception ex)
{
// 2.4 失败：记录日志并自增失败次数，进入下一轮更长的等待
_logger.LogWarning(ex, "重连失败");
Interlocked.Increment(ref _attempt);
}
}
}
finally
{
// 3) 释放重连锁
_reconnectLock.Release();
}
}
5.6. ManagedMqttClient vs bare client
MQTTnet provides two clients: MqttClient (naked) and ManagedMqttClient (managed). Which one you choose is not a matter of taste, but rather a matter of your equipment size, reliability requirements, and depth of customization. Many projects use managed clients to save trouble at the beginning, but later find that they cannot be changed because the key logic is encapsulated internally and cannot be touched.
The hosted version helps you make up for reconnection, offline queue, and subscription recovery; but both “jitter” and “application layer heartbeat protection” must be written by yourself.
suggestion: Small and medium-sized (< 500 devices) use ManagedMqttClient directly to save worry; when large-scale or need to cover the application layer heartbeat, encapsulate the bare client yourself.
⚠️ Version status: MQTTnet 5.x no longer has built-in ManagedMqttClient
from MQTTnet 5.0 At first, the official clearly stated Removal of Managed Client (BREAKING CHANGE) in the upgrade instructions. The managed client capability of MQTTnet.Extensions.ManagedClient is no longer available in the 5.x mainline; the last version with ManagedClient was 4.3.7.1207。
The official words are:
The managed client in version 4 lacks several features which require that the client will be built from scratch. It is recommended to use the regular client and doing the reconnect stuff etc. via your own code.
That is to say:
-
New projects use MQTTnet 5.x: Without CreateManagedMqttClient(), you have to use MqttClient to package a layer of reconnection, offline queue, and resubscription – this is the idea of ReliableMqttClient in this article.
-
Old projects depend on ManagedClient: You can continue to lock MQTTnet 4.3.7.1207, but you will not get the performance optimizations, Memory/Span improvements, and new features of MQTT 5.0 in 5.x.
-
Don’t mix: MQTTnet 5.x + MQTTnet.Extensions.ManagedClient 4.x can’t be programmed basically, and neither the API nor the namespace match up.
So now when talking about “ManagedMqttClient vs bare client”, a more accurate statement is:
In the MQTTnet 5.x era, hosting the client is not a question of “choose or not”, but “you can only build a hosting layer based on the bare client”。
If small and medium-sized projects want to write less code, they can use community encapsulation (such as IResilientMqttClient of MQTTnet.Rx), but for production-level IoT backends, it is recommended to control reconnection/jitter/offline buffering by yourself.
Connection, disconnection callback, and heartbeat are not independent logics, but a closed-loop link that covers each other.
DisconnectedAsync is the first signalhanding over the disconnection fact to the reconnection loop as soon as possible;Heartbeat is the second insuranceused to capture silent disconnections where “the connection appears to be still there, but is actually dead.”
The stability of ClientId determines whether the Broker retains the session during reconnection and whether the subscription can be restored correctly – it is the only unchangeable anchor point in this closed loop.
The following code integrates: exponential backoff + random jitter + re-chaining + re-subscription + application layer heartbeat protection + offline buffering. Can be copied directly into the project.
using Microsoft.Extensions.Logging;
using MQTTnet;
using MQTTnet.Protocol;
using System.Collections.Concurrent;
namespace MqttClient.Demo;
// 可靠 MQTT 客户端。
// 提供六项可靠性能力：指数退避、随机抖动、重连锁、自动重订阅、应用层心跳兜底、离线消息缓冲。
public sealed class ReliableMqttClient : IAsyncDisposable
{
#region 常量
// 默认初始退避秒数。
private const int DefaultBaseDelaySec = 1;
// 默认退避上限秒数。
private const int DefaultMaxDelaySec = 30;
// 默认离线缓冲最大消息数。
private const int DefaultMaxOutboxSize = 10000;
// 默认应用层心跳间隔秒数。
private const int DefaultHeartbeatIntervalSec = 30;
// 单次重连尝试的超时秒数。
private const int ReconnectTimeoutSec = 15;
// 单次心跳等待 PINGRESP 的超时秒数，超时视为心跳失败。
private const int PingTimeoutSec = 5;
#endregion
#region 字段
// 底层 MQTTnet 客户端实例，负责实际的协议收发。
private readonly IMqttClient _client;
// MQTT 连接选项（服务端地址、客户端 ID、保活、CleanSession 等）。
private readonly MqttClientOptions _options;
// 日志记录器，用于输出连接、重连、心跳等运行信息。
private readonly ILogger _logger;
// 订阅登记表：重连后按它自动重订阅。
private readonly ConcurrentDictionary _subscriptions = new();
// 断线期间的待发消息缓冲队列，重连后按 FIFO 顺序补发。
private readonly ConcurrentQueue _outbox = new();
// 重连锁：保证同一时刻只有一个重连循环在跑，避免并发重连。
private readonly SemaphoreSlim _reconnectLock = new(1, 1);
// 全局取消令牌：Dispose 时停掉心跳与重连循环。
private readonly CancellationTokenSource _cts = new();
// 随机数生成器，用于指数退避的抖动计算。
private readonly Random _rand = new();
// 连续失败次数：连接成功时清零，失败时自增，用于计算指数退避。
private int _attempt;
// 指数退避初始秒数。
private readonly int _baseDelaySec;
// 指数退避上限秒数，超过后不再增长。
private readonly int _maxDelaySec;
// 离线缓冲队列最大长度，超出后新消息丢弃。
private readonly int _maxOutboxSize;
// 应用层心跳间隔。
private readonly TimeSpan _heartbeatInterval;
#endregion
#region 事件
// 收到 broker 推送的应用消息时触发。
public event Func? ApplicationMessageReceived;
#endregion
#region 构造函数
// 创建一个可靠 MQTT 客户端实例。
public ReliableMqttClient(
string brokerHost,
int brokerPort,
string clientId,
ILogger logger,
int baseDelaySec = DefaultBaseDelaySec,
int maxDelaySec = DefaultMaxDelaySec,
int maxOutboxSize = DefaultMaxOutboxSize,
int heartbeatIntervalSec = DefaultHeartbeatIntervalSec)
{
// 保存配置
_logger = logger;
_baseDelaySec = baseDelaySec;
_maxDelaySec = maxDelaySec;
_maxOutboxSize = maxOutboxSize;
_heartbeatInterval = TimeSpan.FromSeconds(heartbeatIntervalSec);
// 创建底层客户端
_client = new MqttClientFactory().CreateMqttClient();
// 构建连接选项：固定 clientId + 30s 保活 + CleanSession=false（会话持久化，便于重订阅）
_options = new MqttClientOptionsBuilder()
.WithTcpServer(brokerHost, brokerPort)
.WithClientId(clientId)
.WithKeepAlivePeriod(TimeSpan.FromSeconds(30))
.WithCleanSession(false)
.Build();
// 订阅连接/断线事件
_client.ConnectedAsync += OnConnectedAsync;
_client.DisconnectedAsync += OnDisconnectedAsync;
// 订阅消息接收事件（只注册一次，避免重连后重复挂接导致一条消息被回调多次）
_client.ApplicationMessageReceivedAsync += args =>
{
var msg = args.ApplicationMessage;
return ApplicationMessageReceived?.Invoke(msg) ?? Task.CompletedTask;
};
}
#endregion
#region 公共方法
// 启动客户端：建立连接并启动应用层心跳循环。初始连接失败会自动进入重连流程。
public async Task StartAsync()
{
// 1) 后台启动心跳循环
_ = Task.Run(() => HeartbeatLoopAsync(_cts.Token));
// 2) 尝试首次连接
try
{
await _client.ConnectAsync(_options, _cts.Token);
}
catch (Exception ex)
{
// 3) 初始连接失败：记录日志并触发重连循环
_logger.LogError(ex, "初始连接失败");
_ = ReconnectLoopAsync();
}
}
// 订阅主题。登记到订阅表中，连接建立后立即订阅；断线期间登记的主题会在重连后自动恢复。
public async Task SubscribeAsync(string topic, MqttQualityOfServiceLevel qos)
{
// 1) 登记到订阅表，重连后会按此表恢复
_subscriptions[topic] = qos;
// 2) 已连接则立即下发 SUBSCRIBE
if (_client.IsConnected)
{
await _client.SubscribeAsync(
new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter(topic, qos)
.Build());
}
}
// 发布消息。在线时直接发送；断线或发送失败时进入离线缓冲队列，待重连后自动补发。
public async Task PublishAsync(MqttApplicationMessage message)
{
// 1) 在线则直接发送
if (_client.IsConnected)
{
try
{
await _client.PublishAsync(message);
return;
}
catch (Exception ex)
{
// 2) 在线但发送失败：降级到离线缓冲
_logger.LogWarning(ex, "发布失败，转入离线缓冲");
}
}
// 3) 离线缓冲：队列未满则入队，否则丢弃
if (_outbox.Count < _maxOutboxSize)
{
_outbox.Enqueue(message);
}
else
{
_logger.LogWarning("离线缓冲已满，消息被丢弃");
}
}
// 释放资源：停止心跳与重连循环，断开连接并释放底层客户端。
public async ValueTask DisposeAsync()
{
// 1) 取消所有后台循环
_cts.Cancel();
// 2) 若仍在线则主动断开
if (_client.IsConnected)
{
await _client.DisconnectAsync();
}
// 3) 释放底层资源
_client.Dispose();
_cts.Dispose();
_reconnectLock.Dispose();
}
#endregion
#region 连接事件处理
// 连接成功回调：重置退避计数、恢复订阅、补发离线缓冲消息。
private async Task OnConnectedAsync(MqttClientConnectedEventArgs e)
{
// 1) 重置连续失败计数，下次重连从 base 开始
Interlocked.Exchange(ref _attempt, 0);
_logger.LogInformation("MQTT 连接成功");
// 2) 按订阅表逐个恢复订阅
foreach (var (topic, qos) in _subscriptions)
{
await _client.SubscribeAsync(
new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter(topic, qos)
.Build());
}
// 3) 补发断线期间缓冲的消息
while (_outbox.TryDequeue(out var msg))
{
try
{
await _client.PublishAsync(msg);
}
catch (Exception ex)
{
// 补发失败：重新入队并中断本次补发，等下次连接成功再试
_logger.LogWarning(ex, "补发离线消息失败，重新入队");
_outbox.Enqueue(msg);
break;
}
}
}
// 断线回调：由曾经在线触发的断线才启动重连循环，避免初始连接失败被重复触发。
private Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs e)
{
_logger.LogWarning("MQTT 连接断开：{Reason}", e.Reason);
// 仅当之前确实在线时才触发重连，避免初始连接失败被事件重复触发
if (e.ClientWasConnected)
{
_ = ReconnectLoopAsync();
}
return Task.CompletedTask;
}
#endregion
#region 重连与退避
// 重连循环：通过信号量保证同一时刻只有一个循环；按指数退避 + 随机抖动等待后重试。
private async Task ReconnectLoopAsync()
{
// 1) 非阻塞抢锁：已有人在重连则直接退出
if (!await _reconnectLock.WaitAsync(0))
{
return;
}
try
{
// 2) 循环重试，直到取消或连接成功
while (!_cts.IsCancellationRequested && !_client.IsConnected)
{
// 2.1 计算本次等待时长（指数退避 + 抖动）
var delay = NextDelay();
_logger.LogInformation(
"第 {Attempt} 次重连尝试，等待 {Delay:0.00} 秒",
_attempt,
delay.TotalSeconds);
// 2.2 退避等待，期间被取消则退出
try
{
await Task.Delay(delay, _cts.Token);
}
catch (OperationCanceledException)
{
return;
}
// 2.3 带超时地尝试连接
try
{
using var timeout = new CancellationTokenSource(
TimeSpan.FromSeconds(ReconnectTimeoutSec));
await _client.ConnectAsync(_options, timeout.Token);
// 连接成功，OnConnectedAsync 会重置 _attempt 并做重订阅/补发
return;
}
catch (Exception ex)
{
// 2.4 失败：记录日志并自增失败次数，进入下一轮更长的等待
_logger.LogWarning(ex, "重连失败");
Interlocked.Increment(ref _attempt);
}
}
}
finally
{
// 3) 释放重连锁
_reconnectLock.Release();
}
}
// 计算下一次重连等待时间：base * 2^(n-1)，封顶 max，并乘以 [0.5, 1.0] 的随机抖动。
private TimeSpan NextDelay()
{
// 指数部分：base * 2^n
var n = _attempt + 1;
var cap = Math.Min(_maxDelaySec, _baseDelaySec * Math.Pow(2, n - 1));
// 抖动：0.5 ~ 1.0 之间的随机系数，避免大量客户端同时重连造成惊群
var jitter = 0.5 + _rand.NextDouble() * 0.5;
return TimeSpan.FromSeconds(cap * jitter);
}
#endregion
#region 应用层心跳
// 心跳循环：周期发送 MQTT PINGREQ；失败时主动断开连接，让重连流程接管。
private async Task HeartbeatLoopAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
// 1) 按配置间隔等待
await Task.Delay(_heartbeatInterval, ct);
// 2) 不在线则跳过（重连循环会负责恢复）
if (!_client.IsConnected)
{
continue;
}
// 3) 发送应用层心跳（PINGREQ），带超时避免半开连接永久挂起
bool pong;
try
{
using var pingCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
pingCts.CancelAfter(TimeSpan.FromSeconds(PingTimeoutSec));
pong = await _client.TryPingAsync(pingCts.Token);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
// 超时（非外部取消）：视为心跳失败
pong = false;
}
// 4) 心跳失败：连接可能已半开，强制断开以触发重连
if (!pong)
{
_logger.LogWarning("心跳探测失败，强制重连");
_ = ForceReconnectAsync();
}
}
}
// 主动断开当前连接，触发 DisconnectedAsync 事件进入重连流程。
private async Task ForceReconnectAsync()
{
try
{
// 仅在仍在线时才需要断开
if (_client.IsConnected)
{
await _client.DisconnectAsync();
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "强制断开连接失败");
}
}
#endregion
#region 测试缝（internal，仅用于单元/集成测试）
// 当前连续重连失败次数（测试用）。
internal int AttemptCount => Volatile.Read(ref _attempt);
// 当前离线缓冲队列长度（测试用）。
internal int OutboxCount => _outbox.Count;
// 当前是否处于已连接状态（测试用）。
internal bool IsConnected => _client.IsConnected;
// 当前订阅表的只读快照（测试用）。
internal IReadOnlyDictionary Subscriptions =>
_subscriptions.ToDictionary(kv => kv.Key, kv => kv.Value);
// 计算下一次重连延迟，不实际启动重连（测试用）。
internal TimeSpan NextDelayForTest()
{
return NextDelay();
}
// 手动触发一次重连循环（测试用）。
internal Task TriggerReconnectForTest()
{
return ReconnectLoopAsync();
}
#endregion
}
The previous six sections are all about “what should be done”, and this section is about “what will happen if you don’t do this”. These 5 failures are all pitfalls that I or my team have actually encountered in the production environment – each one corresponds to a certain knowledge point in a previous section. You can think of them as a comparison table of negative teaching materials: if you see any pitfalls, go back and read the solutions in that section.
Five real cases of pitfalls – behind each of them is a configuration discipline that is easily overlooked.
The previous seven sections covered a bunch of principles, strategies, and code, but what you really need before going live is a checklist to check off. These 12 items are the “minimum complete set” that I extracted from all the knowledge points in this article – if they all pass, the connection reliability is basically stable; if one fails, it is a time bomb. It is recommended to print it out and go through the code one by one, rather than relying on memory.
Check items one by one, and pass them all before going online – each item corresponds to a real pitfall.
Friends who are doing MQTT for the Internet of Things can communicate together. I will continue to update this series in this account later, and will also intersperse some records of software exam architect preparation, technology management transformation, and embedded learning.
I am an[IT planner]. I have been working on .NET IoT backend for 8 years. I am moving from pure backend to software and hardware integration architect. The road isn’t easy, but it’s worth it.
Note: The test method code and sample source code have been uploaded to the code library. Those who are interested can take a look.
