3
mica-mqtt 1.0.3 发布,新增 websocket 支持
source link: https://segmentfault.com/a/1190000040519657
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
mica-mqtt 1.0.3 发布,新增 websocket 支持
mica-mqtt 基于 t-io 实现的简单、低延迟、高性能 的 mqtt 物联网开源组件。使用详见 mica-mqtt gitee 源码mica-mqtt-example
模块。
- [x] 支持 MQTT v3.1、v3.1.1 以及 v5.0 协议。
- [x] 支持 websocket mqtt 子协议(支持 mqtt.js)。
- [x] 支持 MQTT client 客户端。
- [x] 支持 MQTT server 服务端。
- [x] 支持 MQTT 遗嘱消息。
- [x] 支持 MQTT 保留消息。
- [x] 支持自定义消息(mq)处理转发实现集群。
- [x] MQTT 客户端 阿里云 mqtt 连接 demo。
- [x] 支持 GraalVM 编译成本机可执行程序。
- [x] 支持 Spring boot 项目快速接入(mica-mqtt-spring-boot-starter)。
- [x] 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();
七、效果演示
八、关注我们
扫描上面二维码,更多精彩内容每天推荐!
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK