FluxMQ—2.0版本更新内容

前言

FLuxMQ是一款基于java开发，支持无限设备连接的云原生分布式物联网接入平台。FluxMQ基于Netty开发，底层采用Reactor3反应堆模型，具备低延迟，高吞吐量，千万、亿级别设备连接；方便企业快速构建其物联网平台与应用.

FluxMQ官网：https://www.fluxmq.com FluxMQ演示系统：http://demo.fluxmq.com/ 。

改动说明

自研EventBus通信组件

研发背景

❝

去除Ignite通信组件，采用自研集群通信，结合分布式订阅树完成性能的大幅度提升 。

❞

Flux1.0版本采用Ignite的message API进行数据路由，此方式主要有以下几个缺点:

1. 集群间通信采用广播方式，大集群下通信性能极低
2. 不支持通配符方式，无法解决通配符的路由
3. 无法实现集群消费能力
4. 对Ignite依赖非常重，导致出现问题排查代价很高，并且无法替换之

EventBus特性

集群通信

基于TCP组件（后面拓展UDP组播等功能），实现集群间启动相互连接，服务端实现端口占用扫描启动：默认48880端口、如果端口占用则依次递增，最大端口号为：49000。 节点启动后，客户端根据配置文件配置的集群IP自动进行端口扫描连接（48880->49000），同时客户端维护与服务端的心跳。避免节点宕机.

报文帧

批量压缩

❝

FluxMQ对于集群消息路由会自动计算TPS，当单节点TPS超过2000时，会自动启动批量压缩功能，以此提高集群间传输性能（对时延要求极高的可以手动关闭批量压缩功能）.

❞

分布式发布订阅匹配树

FluxMQ集群节点间维护一个Root级别的订阅，订阅会分为2种

• 本地订阅
• 远程订阅 为了最快搜索匹配树，订阅信息会维护到每个节点中，当推送Topic通过匹配树匹配后，本地订阅直接Write消息，远程订阅走EventBus系统传输到远程节点。

数据管理

❝

2.0版本新增延迟消息跟会话消息，1.0版本也有保留消息，但是管理页面未实现可视化管理。下面我们介绍下此次改动的一些 。

❞

会话消息

提供分布式会话消息，会话期间Session消息持久化存储，集群宕机后重启，数据不丢失，重启集群后数据重新加载 。

保留消息

根据Topic保留消息，每个TOPIC仅保留一条，当传输的MQTT payload为空时，则清空保留消息。数据持久化存储，重启集群后，数据重新加载 。

延迟消息

❝

FluxMQ提供大批量定时下发Topic指令的能力，单机支持百万级别延迟消息指令下发，在集群模式下，FluxMQ接收到延迟指令后，会自动负载到执行节点执行，当执行节点宕机后，此节点未执行的任务会自动由其他节点继续执行，提供分布式协调任务的能力 。

❞

延迟Topic格式:

                        
                        
                        
                           $DELAY 
                          /延迟秒指/TOPIC
                          

                        
                      

配置持久化

基于Ignite的实现配置数据区持久化，目前持久化的数据内容有以下:

规则引擎

LOG文件打印

❝

此项功能可以用于调试报文，并且于集群各节点生产独立的log文件，用于快速定位问题 。

❞

数据库SQL模板支持Json函数'

数据输入:

                        
                        
                        
                          {
                          

                              
                           "msg" 
                          : {
                          

                                
                           "id" 
                          :
                           "id" 
                          ,
                          

                                
                           "body" 
                          :{
                          

                                  
                           "state" 
                          :1,
                          

                                  
                           "no" 
                          :2
                          

                                }
                          

                              },
                          

                              
                           "messageId" 
                          : 1,
                          

                              
                           "topic" 
                          : 
                           "test" 
                          ,
                          

                              
                           "qos" 
                          : 1,
                          

                              
                           "retain" 
                          : 
                           false 
                          ,
                          

                              
                           "time" 
                          : 
                           "2022 12-22 12:00:00" 
                          ,
                          

                              
                           "clientId" 
                          : 
                           "A1212313" 
                          

                          }
                          

                        
                      

此时我只想插入msg内容下的body结构体，以下是一个通用的插入SQL语句模板:

                        
                        
                        
                          insert into table (clientId,topic,msg) values (
                           '${clientId}' 
                          ,
                           '${topic}' 
                          ,
                           '${json(msg.body)}' 
                          )
                          

                        
                      

通过json(变量名) 方式给结构体转成json字符串替换成插入字段的值 。

多协议模块

目前FluxMQ内置了COAP、WEBSOCKET、I1协议的组件，可以指定端口启动，启动后，可以通过MQTT与协议组件之间交互。每个客户端必须按照FluxMQ的标准进行接入。扩展协议与FluxMQ的MQTT共享以下组件:

• 认证模块
• 规则引擎
• 连接管理
• 日志管理
• 监控管理

上行指令

❝

通过规则引擎配置选择扩展协议数据类型 。

❞

select * from "$EVENT.EXTENSION" 。

传输的数据格式如下:

                        
                        
                        
                          {
                          

                              
                           "protocol" 
                          : 
                           "I1" 
                          ,
                          

                              
                           "cmd" 
                          : 
                           "PUBLISH" 
                          ,
                          

                              
                           "messageId" 
                          : 0,
                          

                              
                           "time" 
                          : 
                           "2023-07-11 21:59:23" 
                          ,
                          

                              
                           "clientId" 
                          : 
                           "clientId" 
                          ,
                          

                              
                           "nodeIp" 
                          : 
                           "127.0.0.1" 
                          ,
                          

                              
                           "clientIp" 
                          : 
                           "127.0.0.1:19999" 
                          ,
                          

                              
                           "body" 
                          : 
                           "body" 
                          

                          }
                          

                        
                      

新增一个转发WEBSOCKET协议的报文

SQL如下:

                        
                        
                        
                          select * from 
                           "  $EVENT  .EXTENSION WHERE protocol='WEBSOCKET'" 
                          

                        
                      

新增一个转发WEBSOCKET 上报协议的报文

SQL如下:

                        
                        
                        
                          select * from 
                           "  $EVENT  .EXTENSION WHERE protocol='WEBSOCKET' AND cmd ='PUBLISH' 
 
                        
                      

下行指令

通过MQTT客户端下发FluxMQ集群指令，即可将指令写给扩展协议客户端，格式如下:

                        
                        
                        
                           $PROTOCOL 
                          /协议名称/{clientId}
                          

                        
                      

连接管理

启动WEBSOCKET协议插件

WEBSOCKET客户端连接

                        
                        
                        
                          ws://123.249.9.130:7777/
                           test 
                          

                        
                      

连接管理