使用开源reactor-netty库，实现MQTT server。集成了springboot autoconfig实现快速注入容器。 框架采用反应式reactor3库，是代码具有低延迟，高吞吐量等特点。
- qos 0,1,2完整实现
- 遗嘱消息,保留消息实现
- 客户端重连机制
- 密码校验,以及版本校验
- 支持ssl加密
- spring容器支持
- channel存储,topic存储,保留消息等外部接口支持
- MQTT 协议同时支持WS/TCP 传输,默认MQTT协议打开WS 8443端口
RsocketServerSession serverSession=TransportServer.create("192.168.100.237",1884)
.auth((s,p)->true)
.heart(100000)
.protocol(ProtocolType.MQTT)
.ssl(false)
.auth((username,password)->true)
.log(true)
.messageHandler(new MemoryMessageHandler())
.exception(throwable -> System.out.println("&&&&&&&&&&&&&&&&&&&&&&&&&&&&"+throwable))
.start()
.block();
serverSession.closeConnect("device-1").subscribe();// 关闭设备端
List<TransportConnection> connections= serverSession.getConnections().block(); // 获取所有链接
RsocketClientSession clientSession= TransportClient.create("127.0.0.1",1884)
.heart(10000)
.protocol(ProtocolType.MQTT) // 指定协议 MQTT 包含 TCP/WS 两个端口 默认WS走的8443 WS协议 仅仅启动TCP协议
.ssl(false) // 开发tls加密
.log(true) // 打印报文日志
.onClose(()->{}) // 客户端关闭事件
.clientId("Comsumer_3") // 客户端id
.password("12") // 密码
.username("123") // 用户名
.willMessage("123") // 遗嘱消息
.willTopic("/lose") // 遗嘱消息topic
.willQos(MqttQoS.AT_LEAST_ONCE) // 遗嘱消息qos
.exception(throwable -> System.out.println("&&&&&&&&&&&&&&&&&&&&&&&&&&&&"+throwable)) // 异常处理
.messageAcceptor((topic,msg)->{
System.out.println(topic+":"+new String(msg));
}) // 消息接收处理
.connect()
.block();
clientSession.sub("test").subscribe(); // 订阅
clientSession.pub("test","Producer_3".getBytes()).subscribe(); // 发布qos0消息
clientSession.pub("test","Producer_1".getBytes(),1).subscribe(); // 发布qos1消息
clientSession.pub("test","Producer_1".getBytes(),true,1).subscribe(); // 发布qos1消息 保留消息
yaml 配置:
iot:
mqtt:
server:
enable: true
host: 192.168.100.237
port: 8081
log: false
protocol: MQTT
heart: 100000
ssl: false
设备校验：实现AuthencationSession接口注入容器即可
异常处理：实现ExceptorAcceptor接口注入容器即可
保留消息处理：实现RsocketMessageHandler接口注入容器即可，默认使用内存。
qq群号： 789331252
