物联网基于Netty主要实战—会话管理

简述

MQTT 的会话是服务端为每个客户端维护的状态信息，用于支持离线消息和持久订阅。至于常见痛点包括，

• 连接频繁断开导致状态丢失
• 离线消息无法及时投递
• 设备重连后订阅列表被清空
• 心跳超时误判为“掉线”

源码

主要概念

• 会话定义：服务端为某个客户端存储的完整状态集合。包括订阅列表、未完成 QoS 消息、在线标识。
• • 多设备共享同一 clientId 时状态被覆盖。
  • 未完成 QoS 消息堆积导致内存泄漏。

会话管理 vs 心跳保活

区别要点：

• 会话管理：跨连接，持久化订阅与离线消息；目标是“断线后记住状态”。老实说，
• 心跳保活：单次连接期间无状态。只传递“我还活着”，目标是“及时发现断线”。
• 触发方式： CONNECT 报文中的 Clean Session 标志决定是否持久化；客户端定时发送 PINGREQ，服务端回复 PINGRESP。
• 结果影响： 若 Clean Session=true，Broker 在重连后恢复会话；若 false，则立即清除。

协同工作流程图

┌─────────────────────────────────────────────────────────┐
│ MQTT 连接生命周期 │
├─────────────────────────────────────────────────────────┤
│ 连接建立 → 正常通信 → 断线 → 重连 │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ 会话加载 心跳维持 会话保存 会话恢复 │
│ │
└─────────────────────────────────────────────────────────┘

直白的理解——银行类比

时间轴 →
第一次连接这方面，你走进银行 → 柜员问“要办什么？”你说“存定期”
→ 银行记下
→ 办理中，柜员每隔30秒问“还在吗？”你答“在”
→ 办完离开，银行保留你的偏好
有钱人给你汇款。银行暂存
第二次连接：你
走进银行 → 柜员调出记录说“记得您，还是存定期吗？说起来，”
→ 柜员把暂存的汇款给你
→ 重新开始每隔30秒问“还在吗？”
⚠️ 痛点提醒：
• 多人共用账号 -> 状态冲突；• 离线期间长时间未投递 -> 消息丢失；话说回来，• 心跳过短 -> 假掉线。

源码实现细节 – Server 与 Client 分别说明

Server 实现

package com.jysemel.iot;import com.jysemel.iot.handler.MqttSessionHandler;import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.codec.mqtt.MqttDecoder;import io.netty.handler.codec.mqtt.MqttEncoder;public class MqttSessionServer {
public static void main throws Exception {
EventLoopGroup boss = new NioEventLoopGroup;EventLoopGroup worker = new NioEventLoopGroup;try {
ServerBootstrap b = new ServerBootstrap;b.group
.channel
.childHandler {
@Override protected void initChannel {
ChannelPipeline p = ch.pipeline;p.addLast)，p.addLast;p.addLast)，// 主要业务逻辑
}
});b.bind.sync.channel.closeFuture.sync;其实，} finally {
boss.shutdownGracefully;worker.shutdownGracefully;}
}
}

P.S. 如果出现 “地址已占用” 错误，请检查端口是否被其他 MQTT Broker 占用。说起来，

Client 实现

package com.jysemel.iot;import com.jysemel.iot.handler.TestMqttClientHandler;import io.netty.bootstrap.Bootstrap;怎么说呢，import io.netty.buffer.Unpooled;import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;import io.netty.handler.codec.mqtt.*;import java.nio.charset.StandardCharsets;说起来，public class TestMqttClient {
private Channel channel;private String clientId;private int messageId = 1;public void connect throws Exception {
this.clientId = clientId;EventLoopGroup group = new NioEventLoopGroup;Bootstrap b = new Bootstrap;b.group
.channel
.handler {
@Override protected void initChannel {
ChannelPipeline p = ch.pipeline;p.addLast)，p.addLast;p.addLast)，}
});ChannelFuture f = b.connect.sync;其实，channel = f.channel;// 构造 CONNECT 报文
MqttConnectMessage connectMsg = MqttMessageBuilders.connect
.clientId
.cleanSession
.keepAlive
.protocolVersion
.build;channel.writeAndFlush;System.out.println;}
public void subscribe {
MqttSubscribeMessage subscribeMsg =
MqttMessageBuilders.subscribe
.messageId
.addSubscription,topic)
.build;channel.writeAndFlush;其实，System.out.println;}
public void publish {
MqttPublishMessage publishMsg =
MqttMessageBuilders.publish
.topicName
.qos)
.messageId
.payload))
.build;channel.writeAndFlush;System.out.println;}
public void disconnect {
channel.writeAndFlush.build);System.out.println;channel.close;}
public static void main throws Exception {
TestMqttClient client = new TestMqttClient;// 示例：不同 clientId / cleanSession 的组合演示效果
String clientId1 = "client-A";
boolean cleanA = false;
// 持久化 session
client.connect;Thread.sleep;// 等待握手完成
client.subscribe;// QoS=1 保证一次投递
Thread.sleep;client.publish;Thread.sleep;client.disconnect;// 模拟网络掉线
Thread.sleep;// 等待服务器确认掉线并保存 state
//
连上，同一 deviceid。验证 session 恢复与离线消息推送
client.connect;Thread.sleep;// 此时应该能收到之前发布但未确认的 QoS=1 消息
System.exit;}
}

TIPS: ① 若出现 “NoSuchElementException” 或 “IndexOutOfBoundsException”。请检查 MESSAGEID++/--  — ② 为防止内存泄漏，请在服务器关闭前执行 #channel.close;③ 多台设备共用同一 Client ID 时可通过加前缀区分或使用独立 Broker。