简介基于 Java AIO 实现的低延迟、高性能百万级 MQTT 客户端组件与 Broker 服务定位为物联网、边缘计算场景中需要自建消息服务器的 Java 开发者解决多协议接入、高并发连接和集群扩展等核心问题。压缩包共 282 个文件文件类型以 221 个 Java 源码为主体含 15 个 Markdown 文档、13 个 XML 和 11 个 YAML 配置、3 个 JSON 及 HTTP 调试脚本等配套较完整整体仅 502KB目录紧凑适合直接导入工程研读。完整支持 MQTT v3.1/v3.1.1/v5.0、WebSocket MQTT 子协议、遗嘱与保留消息同时提供 REST API、GraalVM 本机编译、PrometheusGrafana 监控和 Redis Pub/Sub 集群方案方便从单机接入平滑演进到分布式部署。内容还覆盖 mica-mqtt-api.http、MqttDecoder/MqttEncoder 等关键实现便于理解协议编解码、Spring Boot 快速接入、阿里云 MQTT 连接示例以及自定义消息处理转发的集群机制。已有 279 人学习下载对希望掌握高性能 MQTT 组件设计、集群搭建与监控运维的开发者具有较高参考价值。1. 基于 Java AIO 的 MQTT 客户端与服务端百万连接场景下的选型与落地如果你维护过物联网平台的接入层大概率遇到过这类尴尬用 Netty 写的 MQTT Broker 在几千连接时风平浪静压到十万连接就开始频繁 Full GC线程模型一调再调最后还是靠加机器硬扛。这个基于 Java AIO异步 I/O实现的 mqtt client 和 broker 组件走的是另一条路——它把 IO 线程压到个位数靠操作系统级别的异步通知处理海量连接实测在大规模长连接场景下能撑到百万级。组件同时支持 MQTT v3.1、v3.1.1、v5.0还带 websocket 子协议、HTTP REST API、遗嘱消息、保留消息、基于 Redis pub/sub 的集群方案甚至能通过 GraalVM 编译成本地镜像。这篇笔记我会从选型理由讲起把 client 端接入、broker 端部署、集群配置、踩坑记录和监控验证逐层拆开适合正在做 IoT 接入层选型或想优化现有 MQTT 服务的 Java 工程师参考。需要说明的是我关注的是落地路径——代码怎么跑起来、参数怎么调、哪些坑必须绕开。2. Java AIO 与 MQTT 协议实现百万连接的架构逻辑2.1 为什么选 AIO 而不是 NIO从 C10K 到 C1000K 的思维转变做 Java 网络编程的人最熟悉的是 NIO基于 Selector 的事件轮询模型。Netty 在 NIO 上做了大量优化但本质上仍然是「一个或几个线程轮询所有 channel 的事件」。连接数上来之后每次 select 返回的 selectedKeys 数量巨大遍历和处理这些 key 本身就消耗 CPU。而且 NIO 的读写操作在多数情况下仍然是同步的业务线程需要等待 IO 完成。Java AIOAsynchronousSocketChannel、AsynchronousServerSocketChannel把读写回调直接交给操作系统应用层注册 CompletionHandler 即可读写完成时系统调用回调。这个组件的作者正是因为看中了 AIO 在纯异步读写下对线程资源的释放才决定基于 AIO 实现 MQTT 编解码与消息分发。常见做法是设置一个较小的 IO 线程池比如 CPU 核数每个线程处理大量连接的异步事件真正做到了「连接百万线程几十」。这里需要注意的是AIO 在 Linux 底层依赖 epoll 的 ET 模式在 Windows 上则是 IOCP两端的线程模型差异会导致某些隐藏 bug后文避坑会提到。2.2 协议栈分层从 MqttDecoder 到消息路由项目源码里能看到MqttDecoder.java、MqttEncoder.java、DefaultMessageSerializer.java这几个关键类。它们共同构成了协议栈的三层解码层负责把字节流解析成 MQTT 报文编码层负责把响应报文写回Serializer 则负责将消息载荷与 Java 对象互转。解码层直接面对 TCP 粘包拆包这里用 AIO 的 ByteBuffer 配合自定义的 MQTT 报文长度解析算法处理方式与 Netty 的 ByteToMessageDecoder 类似但因为是异步回调需要自己管理半包状态的缓存。核心逻辑是先读固定头解析剩余长度再根据报文类型读取可变头与载荷解析到完整的 MQTT 消息后交给上层处理。// 伪代码示意AIO 通道读取回调中的半包处理 private void onReadCompleted(Integer result, AsynchronousSocketChannel channel, ByteBuffer buffer) { buffer.flip(); // 检查当前 buffer 中是否有完整的 MQTT 报文 while (buffer.remaining() 0) { int mark buffer.position(); MqttMessage message decoder.decode(buffer); // 从 MqttDecoder 中解析 if (message null) { buffer.position(mark); // 半包恢复位置等待下次读取 break; } handleMessage(message, channel); } buffer.compact(); channel.read(buffer, buffer, this); // 继续异步读取 }这段代码的核心是decoder.decode(buffer)返回 null 时把 position 恢复到解析前的位置等待剩余字节到达。因为 AIO 每次读取不一定包含完整报文所以必须保留半包数据。buffer.compact()用于把未读完的数据挪到 buffer 头部避免覆盖。建议初始化 ByteBuffer 时用Buffer.allocateDirect堆外内存能减少一次内核到 JVM 堆的拷贝对吞吐提升明显。参数上读取 buffer 大小建议设置成 4096 到 8192 字节太小会导致频繁读取太大浪费内存。2.3 百万连接下的小心思过期会话与心跳保活MQTT 协议要求 client 定期发送 PINGREQbroker 返回 PINGRESP。但这个组件在处理心跳时有个特点它不单独为每个连接维护 Timer 定时器而是用「最后活跃时间戳 统一扫描」。每次收到任何报文都刷新时间戳broker 的后台线程每隔一个心跳周期扫描所有会话把超过keepalive * 1.5的连接判定为死连接并清理。这个设计在百万连接场景下非常重要因为每个连接一个 Timer 意味着几百万个 Timer 对象压在 JVM 堆里GC 会立刻成为瓶颈。// 服务端心跳扫描的简化逻辑 public void checkAlive() { long now System.currentTimeMillis(); for (ClientSession session : sessionManager.getAllSessions()) { long lastAlive session.getLastAliveTime(); int keepalive session.getKeepAliveSeconds(); if (now - lastAlive keepalive * 1.5L 1000) { closeSession(session, heartbeat timeout); } } }扫描周期建议设置为 5 秒一次*1.5是常见的容忍系数。如果你接入的是 485 透传网关这类设备它们的心跳间隔可能不标准keepalive * 1.5不够时可以把系数调大或直接配置为固定值代价是断线检测变慢。另一个隐藏点sessionManager 若用 ConcurrentHashMap 存储所有会话百万连接下 map 的遍历开销不可忽视这个组件内部用的是分段结构性能尚可。3. MQTT Client 客户端实战从连接建立到消息收发3.1 快速接入用 Maven 依赖拉起 client这个组件的 client 端支持 Spring Boot 自动配置官方推荐在pom.xml中引入 starter。我没有写死版本号因为项目还在活跃迭代你使用时以仓库当前 release 为准。dependency groupIdnet.dreamlu/groupId artifactIdmica-mqtt-client-spring-boot-starter/artifactId version${mica-mqtt.version}/version /dependency引入后在application.yml中做基础配置。这里的连接参数有几个值得注意uri支持tcp://和ws://前缀如果给 485 设备做透传一般用 tcp如果对接浏览器端 mqtt.js则必须用 ws。client-id建议在设备端由设备唯一标识生成便于后续踢下线或追踪。username和password在 MQTT v3.1.1 中是可选的但云平台要求必须携带。mica: mqtt: client: enabled: true uri: tcp://localhost:1883 client-id: ${random.uuid} username: admin password: 123456 timeout: 10 keep-alive: 60keep-alive设置成 60 秒意味着服务端和客户端都会在 90 秒内没有消息时主动断开。如果你通过 485 网关采集数据上报频率可能超过 60 秒建议将 keep-alive 调大到 300避免中间链路误判断线。3.2 订阅与发布回调消息时别做耗时操作订阅和发布是 MQTT 最核心的操作。组件提供了IMqttClient接口注入后直接调用subscribe和publish。下面这段代码演示了订阅一个主题并在回调中处理消息。Autowired private IMqttClient mqttClient; public void subscribeAndHandle() { mqttClient.subscribe(device//status, (topic, payload) - { // 注意这里在 IO 线程中回调禁止阻塞 String json new String(payload, StandardCharsets.UTF_8); System.out.println(topic: topic , payload: json); // 应该把消息丢给线程池处理 businessExecutor.execute(() - { handleDeviceStatus(json); }); }); }回调函数的参数topic支持 MQTT 通配符匹配代表一层通配#代表多层通配。在这个组件中回调执行在内部 IO 线程上如果你在回调里做数据库写入、Redis 操作或远程调用会阻塞后续所有消息的读取。我一般会在回调里立刻把消息丢进一个独立的业务线程池线程池大小根据设备量定常见做法是核心线程数 4最大线程数 8队列容量 10000。需要注意的是组件发消息时如果客户端不在线默认会丢弃除非你开启了 Qos1 且设置会话保存这是 MQTT 语义的一部分。3.3 遗嘱消息与保留消息设备掉线的最后一声IoT 场景里设备异常断电是最常见的。若不做遗嘱处理服务端直到心跳超时才会感知设备下线期间业务方可能还在往这个设备发指令。这个组件支持在连接时设置遗嘱消息设备异常断开后 broker 会立即代为发布遗嘱到指定主题。MqttProperties properties new MqttProperties(); properties.setWillTopic(device/offline); properties.setWillMessage(device-123-down); properties.setWillQos(1); mqttClient.connect(properties);遗嘱的作用在于你的监控平台可以订阅device/offline主题实时感知设备异常而不是等心跳超时。保留消息则不同它让 broker 保存某个主题最后一条消息新订阅者连接后立刻收到这条消息适合传递设备当前状态或配置。注意这个组件中遗嘱消息和保留消息默认都是关闭的必须显式设置。我在项目里踩过坑设备正常断开时也会触发遗嘱需要在业务侧判断 disconnected 原因再决定是否报警。4. 搭建 MQTT Broker 服务端从单机到 Redis 集群4.1 单机部署一行命令启动 broker这个组件既能当 client 也能当 broker。单机模式下直接在 Java 进程里 new 一个MqttServer即可也支持 Spring Boot starter。部署时最关心三个端口MQTT TCP 端口默认 1883、WebSocket 端口默认 8083、HTTP Rest API 端口默认 8082。java -jar mica-mqtt-server.jar \ --mqtt.port1883 \ --websocket.port8083 \ --http.port8082 \ --mqtt.usernameadmin \ --mqtt.password123456启动日志里会显示实际绑定的端口。如果是在云服务器上部署记得安全组放行这三个端口否则外部设备永远连不上。--mqtt.username与--mqtt.password设置后所有 client 连接都必须携带相同凭证不推荐在公网环境不设密码因为 MQTT 协议本身是明文传输密码会被抓包看到。4.2 HTTP Rest API端到端的管理入口这个组件提供了 HTTP API 用于查看连接信息和发布消息。对于运维和业务系统来说非常实用。常见端点包括GET /api/mqtt/clients获取在线客户端列表POST /api/mqtt/publish发送消息。调用时注意鉴权方式默认情况下这些接口没有鉴权生产环境必须在内网或加一层网关。# 查看在线客户端 curl -X GET http://localhost:8082/api/mqtt/clients # 向指定主题发布消息 curl -X POST http://localhost:8082/api/mqtt/publish \ -H Content-Type: application/json \ -d {topic:device/001/command, payload:reboot, qos:1}qos参数可选 0、1、2。qos1 能保证消息至少到达一次但可能重复业务侧要做幂等。如果是对 485 设备下发指令通常用 qos0 即可因为设备应答更快重复指令反而会导致设备执行两次。HTTP API 返回的 JSON 结构里success字段判断是否发送成功message字段携带失败原因比如 client 不存在或主题没有订阅者。4.3 基于 Redis Pub/Sub 的集群消息转发的横向扩展单机 broker 撑不住百万连接时最常见的方案是对 broker 做集群。这个组件没有采用复杂的 Raft 协议而是结合 Redis Pub/Sub 实现节点间的消息转发。当某台 broker 收到一条发布到特定主题的消息它会将该消息通过 Redis 广播给集群内其他 broker再由其他 broker 转发给各自连接的订阅者。// 伪代码示意Redis 消息订阅与本地转发 public void onRedisMessage(String channel, String message) { MqttPublishMessage publishMsg JSON.parseObject(message, MqttPublishMessage.class); // 在本地 broker 的会话中查找订阅者 ListClientSession sessions sessionManager.match(publishMsg.getTopic()); for (ClientSession session : sessions) { session.writeMessage(publishMsg); } }实现细节上Redis 的 channel 名称通常约定为集群主题比如mqtt-cluster-topic。消息体里包含原始主题、载荷、Qos、保留标志等。这里真正要处理的是消息的循环转发问题——broker A 从 Redis 收到消息后不应该再把消息重新发回 Redis。这个组件的做法是在消息体里携带来源 broker 编号收到消息后判断来源是否等于自己相等则直接丢弃。集群的拓扑结构是星型每个 broker 连接同一个 Redis 实例。如果 Redis 成为瓶颈边缘场景下可以先考虑 Redis Cluster而不是直接改架构。5. 避坑指南从 AIO 到 MQTT 的五个血泪教训5.1 AIO 读回调里的零拷贝陷阱现象压测时吞吐量上不去CPU 占用却居高不下用jstack看到大量线程阻塞在Unsafe.copyMemory。原因我在初期用heap buffer接收 AIO 数据然后又把数据拷贝到另一个 byte[] 做协议解析。AIO 回调本来就在堆外内存中写入数据再复制到堆内导致不必要的内存拷贝。这个组的MqttDecoder直接基于ByteBuffer解析原则上要保持 buffer 中的数据处理链路统一。解决使用ByteBuffer.allocateDirect()分配堆外内存并在解码时直接从buffer切片获取数据尽量不要把get(byte[])的结果再重复组装。如果必须传给业务线程建议发送前copy一次并同步回收。5.2 keep-alive 参数不一致导致的随机断连现象部分 485 透传网关连接后 3 分钟内必然掉线有的网关掉线后自动重连但频繁掉线导致消息丢失。原因网关设备的 MQTT 协议栈实现不标准它发送的 PINGREQ 周期与 broker 端配置的 keep-alive 不一致。比如 broker 设置了 60 秒网关内部实际 90 秒才发心跳broker 按1.5*6090秒判定超时此时网关的心跳刚刚发出临界点上被误杀。解决先用 Wireshark 抓包确认网关真实心跳周期再把 broker 的 keep-alive 容忍系数调成 2 或单独为该网关指定更长的 keep-alive。另外确认网关是否发送了 0 心跳keepalive00 代表服务端不检测这种设备不能直接接入公网 broker容易成为僵尸连接。5.3 遗嘱消息误报正常断开也触发遗嘱现象平台监控经常收到「设备离线」报警但实际设备只是主动重启或远程升级。原因服务端在检测到 TCP 断开时无论断开原因是异常掉线还是客户端正常发送 DISCONNECT 报文后关闭都会触发遗嘱。我最初在服务端 handler 里统一调用了遗嘱发布逻辑。解决在连接关闭时检查对方是否发送了 DISCONNECT 报文。如果收到 DISCONNECT则标记该连接为「优雅退出」不发布遗嘱反之才触发遗嘱。组件默认对正常 disconnect 不发遗嘱但需要你在业务层处理遗嘱的发布时机避免在channelInactive里无脑发布。5.4 百万连接下的文件描述符上限现象压测到 30 万连接时新客户端连接直接被拒绝日志中报Too many open files。原因Linux 系统默认ulimit -n是 1024即单进程只能打开 1024 个文件描述符。MQTT 长连接每个 TCP 连接占用一个 fd即便是 AIO 也逃不开系统限制。解决修改/etc/security/limits.conf将nofile设成 1048576并且确认客户端连接的线程数不大于系统限制。修改后需要重启 Java 进程。此外/etc/sysctl.conf中将net.ipv4.tcp_max_syn_backlog调大避免内核丢包。# /etc/security/limits.conf * soft nofile 1048576 * hard nofile 10485765.5 Redis 集群消息积压导致的消息乱序现象集群模式下设备上报的消息偶尔出现乱序比如后发送的消息先被业务处理。原因Redis Pub/Sub 本身不保证消息的顺序性实际上 Redis 单实例是保序的但多个 broker 节点同时写入 Redis 消息时不同 broker 转发到同一个业务消费者时顺序无法保证。更常见的原因是业务侧用了多线程处理同主题消息导致乱序。解决如果业务对顺序有要求如设备状态机的变更在订阅回调里按设备 ID 做哈希分发到同一个处理线程。String.format(device-%s, deviceId)作为 key用ConcurrentHashMap维护每个 deviceId 对应的单线程 Executor从根上消除并发乱序。6. 验证与进阶从压测脚本到 Prometheus 监控真正验证一个 MQTT broker 能否支撑百万连接不能只靠口头说。我通常分三步走先启动 broker 实例用这个组件自带的 client 端写一个压测程序模拟多设备连接与消息收发再对接 Prometheus Grafana 观察指标最后用 GraalVM 编译成本地镜像部署到边缘网关验证资源占用。压测程序的思路很简单创建 N 个 client 实例每个 client 连接 broker 后订阅自己的主题并周期发布消息。这里有个常见误区如果每台机器只开一个进程创建十万个 client文件描述符和线程栈会率先耗尽。正确做法是拆到多台压测机每台进程内创建一万个 client并通过System.nanoTime()统计消息端到端延迟。// 压测创建连接的简化代码 for (int i 0; i 10000; i) { IMqttClient client MqttClient.create() .serverAddr(tcp://192.168.1.10:1883) .clientId(stress- i) .connect(); client.subscribe(test/ i, (topic, bytes) - { long now System.nanoTime(); // 与消息内嵌入的时间戳计算延迟 }); clients.add(client); }连接白屏后观察 broker 进程的内存与 GC 情况。这个组件因为 IO 线程少正常情况下 GC 压力远小于 Netty但它的内存占用大头在消息缓冲上。如果每台压测机出现大量ReadPendingException说明 AIO 读请求积压可能是写频太快压过了解码速度此时要调整每次 publish 的间隔或者增大读缓冲。监控方面组件暴露了 Prometheus 端点默认路径是/actuator/prometheus或自定义的 metrics 接口。关键指标包括连接数、消息吞吐、消息积压数。在 Grafana 里我通常建一个 API 面板包含四个图表在线连接数每小时趋势、每秒消息量topic 维度的 TOP10、遗嘱触发次数、redis 转发延迟。这里有一个指标要特别关注——「连接建立失败次数」如果这个值大于 0说明系统文件描述符或线程池达到上限往往是容量规划的第一信号。关于 GraalVM 编译。这个组件宣称支持编译成本地镜像我试过一次确实可以。核心坑在于netty类库和反射的使用。如果你要编译 native image需要在配置文件中显式声明反射类尤其是 MQTT 消息序列化类。否则启动时会报ClassNotFoundException。我在边缘网关的容器里直接跑本地镜像内存占用从 300MB 降到 80MB这对于嵌入式设备非常关键。但要注意本地镜像首次启动需要 1-2 秒初始化比 JVM 慢因为 Graal 需要预留堆内存。最后一个建议无论你用它做 client 还是 broker一定要在正式上压测前开启 GC 日志和 AIO 线程的堆栈日志。我经历过一次诡异的连接假死表面现象是 broker 不响应但进程还活着最后通过jstack -F发现 Aio 回调线程被 full GC 停顿卡住这是 JVM 与 AIO 交互时的常见问题。从那以后我每次做百万连接验证都会强制走一遍「先压测-再监控-后调参」的流程并同时记录 GC 日志和netstat -s的断线统计。希望帮到你少踩这三个数字的坑。本文还有配套的精品资源点击获取