mica-mqtt 1.0.3 发布，新增 websocket 支持
一、简介 mica-mqtt 基于 t-io 实现的简单、低延迟、高性能 的 mqtt 物联网开源组件。使用详见 mica-mqtt gitee 源码mica-mqtt-example 模块。 二、功能 支持 MQTT v3.1、v3.1.1 以及 v5.0 协议。 支持 websocket mqtt 子协议（支持 mqtt.js）。 支持 MQTT client 客户端。 支持 MQTT server 服务端。 支持 MQTT 遗嘱消息。 支持 MQTT 保留消息。 支持自定义消息（mq）处理转发实现集群。 MQTT 客户端 阿里云 mqtt 连接 demo。 支持 GraalVM 编译成本机可执行程序。 支持 Spring boot 项目快速接入（mica-mqtt-spring-boot-starter）。 mica-mqtt-spring-boot-starter 支持对接 Prometheus + Grafana。 三、待办 优化处理 mqtt session，以及支持部分 mqtt v5.0 新特性。 四、更新记录 ✨mica-mqtt server 添加 websocket mqtt 子协议支持（支持 mqtt.js）。 ✨mica-mqtt server ip，默认为空，可不设置。 ✨mica-mqtt client去除 CountDownLatch 避免启动时未连接上服务端卡住。 ✨mica-mqtt client 添加最大包体长度字段，避免超过 8092 长度的包体导致解析异常。 ✨mica-mqtt client 添加连接监听 IMqttClientConnectListener。 ✨mica-mqtt 3.1 协议会校验 clientId 长度，添加配置项 maxClientIdLength。 ✨mica-mqtt 优化 mqtt 解码异常处理。 ✨mica-mqtt 日志优化，方便查询。 ✨mica-mqtt 代码优化，部分 Tio.close 改为 Tio.remove。 ✨mica-mqtt-spring-boot-example 添加 Dockerfile，支持spring-boot:build-image。 ✨完善 mica-mqtt-spring-boot-starter，添加遗嘱消息配置。 ⬆️ 升级 t-io 到 3.7.4。 五、Spring boot 快速接入 5.1 添加依赖 <dependency> <groupId>net.dreamlu</groupId> <artifactId>mica-mqtt-spring-boot-starter</artifactId> <version>1.0.3</version> </dependency> 5.2 服务端配置示例 mqtt: server: enabled: true # 是否开启，默认：true ip: 127.0.0.1 # 服务端 ip 默认：127.0.0.1 port: 5883 # 端口，默认：1883 name: Mica-Mqtt-Server # 名称，默认：Mica-Mqtt-Server buffer-allocator: HEAP # 堆内存和堆外内存，默认：堆内存 heartbeat-timeout: 120000 # 心跳超时，单位毫秒，默认: 1000 * 120 read-buffer-size: 8092 # 接收数据的 buffer size，默认：8092 max-bytes-in-message: 8092 # 消息解析最大 bytes 长度，默认：8092 debug: true # 如果开启 prometheus 指标收集建议关闭 websocket-enable: true # 开启 websocket 子协议，默认开启 websocket-port: 8083 # websocket 端口，默认：8083 5.3 服务端可实现接口（注册成 Spring Bean 即可） 接口 是否必须 说明 IMqttServerAuthHandler 是 用于客户端认证 IMqttMessageListener 是 消息监听 IMqttConnectStatusListener 是 连接状态监听 IMqttSessionManager 否 session 管理 IMqttMessageStore 集群是，单机否 遗嘱和保留消息存储 AbstractMqttMessageDispatcher 集群是，单机否 消息转发，（遗嘱、保留消息转发） IpStatListener 否 t-io ip 状态监听 5.4 服务端自定义配置（可选） @Configuration(proxyBeanMethods = false) public class MqttServerCustomizerConfiguration { @Bean public MqttServerCustomizer activeRecordPluginCustomizer() { return new MqttServerCustomizer() { @Override public void customize(MqttServerCreator creator) { // 此处可自定义配置 creator，会覆盖 yml 中的配置 System.out.println("----------------MqttServerCustomizer-----------------"); } }; } } 5.5 MqttServerTemplate 使用示例 import net.dreamlu.iot.mqtt.codec.MqttQoS; import net.dreamlu.iot.mqtt.spring.server.MqttServerTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.nio.ByteBuffer; /** * @author wsq */ @Service public class ServerService { @Autowired private MqttServerTemplate server; public boolean publish(String body) { server.publishAll("/test/123", ByteBuffer.wrap(body.getBytes())); return true; } } 5.6 基于 mq 消息广播集群处理 实现 IMqttConnectStatusListener 处理设备状态存储。 实现 IMqttMessageListener 将消息转发到 mq，业务按需处理 mq 消息。 实现 IMqttMessageStore 存储遗嘱和保留消息。 实现 AbstractMqttMessageDispatcher 将消息发往 mq，mq 再广播回 mqtt 集群，mqtt 将消息发送到设备。 业务消息发送到 mq，mq 广播到 mqtt 集群，mqtt 将消息发送到设备。 5.7 Prometheus + Grafana 监控对接 得益于 t-io 良好的设计，监控指标直接对接的 t-iostat，目前支持下列指标，后期会不断完善。 支持得指标 说明 mqtt_connections_accepted 共接受过连接数 mqtt_connections_closed 关闭过的连接数 mqtt_connections_size 当前连接数 mqtt_messages_handled_packets 已处理消息数 mqtt_messages_handled_bytes 已处理消息字节数 mqtt_messages_received_packets 已接收消息数 mqtt_messages_received_bytes 已处理消息字节数 mqtt_messages_send_packets 已发送消息数 mqtt_messages_send_bytes 已发送消息字节数 关于 mica-mqtt-spring-boot-starter 更多请查看文档：https://gitee.com/596392912/mica-mqtt/tree/master/mica-mqtt-spring-boot-starter 六、普通 java 项目接入 6.1 maven 依赖 <dependency> <groupId>net.dreamlu</groupId> <artifactId>mica-mqtt-core</artifactId> <version>1.0.3</version> </dependency> 6.2 mica-mqtt 客户端 // 初始化 mqtt 客户端 MqttClient client = MqttClient.create() .ip("127.0.0.1") .port(1883) // 默认：1883 .username("admin") .password("123456") .version(MqttVersion.MQTT_5) // 默认：3_1_1 .clientId("xxxxxx") // 默认：MICA-MQTT- 前缀和 36进制的纳秒数 .connect(); // 连接 // 消息订阅，同类方法 subxxx client.subQos0("/test/#", (topic, payload) -> { logger.info(topic + '\t' + ByteBufferUtil.toString(payload)); }); // 取消订阅 client.unSubscribe("/test/#"); // 发送消息 client.publish("/test/client", ByteBuffer.wrap("mica最牛皮".getBytes(StandardCharsets.UTF_8))); // 断开连接 client.disconnect(); // 重连 client.reconnect(); // 停止 client.stop(); 6.3 mica-mqtt 服务端 // 注意：为了能接受更多链接（降低内存），请添加 jvm 参数 -Xss129k MqttServer mqttServer = MqttServer.create() // 默认：127.0.0.1 .ip("127.0.0.1") // 默认：1883 .port(1883) // 默认为： 8092（mqtt 默认最大消息大小），为了降低内存可以减小小此参数，如果消息过大 t-io 会尝试解析多次（建议根据实际业务情况而定） .readBufferSize(512) // 自定义认证 .authHandler((clientId, userName, password) -> true) // 消息监听 .messageListener((clientId, topic, mqttQoS, payload) -> { logger.info("clientId:{} topic:{} mqttQoS:{} message:{}", clientId, topic, mqttQoS, ByteBufferUtil.toString(payload)); }) // ssl 配置 .useSsl("", "", "") // 自定义客户端上下线监听 .connectStatusListener(new IMqttConnectStatusListener() { @Override public void online(String clientId) { } @Override public void offline(String clientId) { } }) // 自定义消息转发，可用 mq 广播实现集群化处理 .messageDispatcher(new IMqttMessageDispatcher() { @Override public void config(MqttServer mqttServer) { } @Override public boolean send(Message message) { return false; } @Override public boolean send(String clientId, Message message) { return false; } }) .debug() // 开启 t-io debug 信息日志 .start(); // 发送给某个客户端 mqttServer.publish("clientId","/test/123", ByteBuffer.wrap("mica最牛皮".getBytes())); // 发送给所有在线监听这个 topic 的客户端 mqttServer.publishAll("/test/123", ByteBuffer.wrap("mica最牛皮".getBytes())); // 停止服务 mqttServer.stop(); 七、效果演示 建议关注如梦技术码云：https://gitee.com/596392912，更多优秀开源项目等你来发现。
