通常我们使用 Mqtt Broker 来实现 mqtt 相关客户端之间的通信，我们可以安装 EMQ 或使用类似阿里提供的 MQTT 代理服务，不管是客户端还是服务端，对于 Broker来说都是 Mqtt Client。
但是在某些场景下还是需要实现自己的 MQTT Server，比如笔者经常需要同时管理几十几百个设备甚至上千个设备，很多都是私有部署，如果每次部署都需要搭一个 Broker 比较重，这边就是一个普通的 Java Spingboot 后台服务，启动就能简单管理多个设备。
所以基于 Netty 实现了这个 MQTT Server，作为一个库，在很多项目上用到了，目前使用比较稳定，没有做专门的压力测试，只是用 python 模拟了2000个设备来同时连接一个服务，是没有问题的。
Netty 用处很广，笔者开源的这个项目可以作为大家使用 Netty 的参考。代码比较简洁，源文件很少。
源码已提交到 Github。主要说明参考如下：
一. 功能概述
- 启动一个MQTT Server实例监控链接上的 MQTT Client
- 开发者只需要实现对应的接口来处理在线、离线、消息处理等事件。无需考虑内部实现细节。
- 可以实时获取到所有在线 MQTT Client 的列表和信息
- 可以利用和特定Client链接的通道来接收和发送数据
- test目录里是用python模拟2000个mqtt客户端来同时链接
二. 使用方式
1. 修改build.gradle增加依赖
可以通过依赖我们内部的企业的库中心来获取最新的版本。当然你可以下载源码加入到自己的项目里
......
repositories {
maven { url 'https://maven.aliyun.com/repository/public/' }
mavenCentral()
maven {
allowInsecureProtocol=true
url 'http://118.253.150.123:8081/repository/d1-java/'
}
}
dependencies {
implementation('d1.duoxian:mqttserver:2.0.7')
......
}
......
2. 启动 MQTT Server 服务
MqttServerService mqttserver = new MqttServerService();
//1)简单启动服务，其中60001是监听的端口，listner是实现 `interface IMqttMessageListener`接口的对象，后面会提到
mqttserver.startup(60001,listener,null);
//2）启动参数还有很多个，大部分都可以用缺省值，完整参数参考MqttServerServiceOption
mqttserver.startup(new MqttServerServiceOption.Builder() //需要改那个参数就设置那个，大部分有缺省值，不需要设置
.port(port)
.messageListener(listener)
.verifyListener(verifyListener)
.checkOfflineInterval(checkOfflineInterval)
.maxBytesInMessage(maxBytesInMessage)
.build());
//3）以下是参数列表
/**
* 缺省的字符集，用于把mqtt发送接收的数据二进制转换成string，缺省是StandardCharsets.UTF_8
*/
private Charset defaultCharset = StandardCharsets.UTF_8;
/**
* 监听的端口
*/
private Integer port;
/**
* 消息监听的接口实现实例
*/
private IMqttMessageListener messageListener;
/**
* 校验客户端身份的接口实现实例，如果不校验用户密码传null即可
*/
private IMqttVerifyListener verifyListener;
/**
* 负责接收accept消息的线程数，通常1个线程(缺省）即可，传0或负数则为1，大于5的数则为5
*/
private Integer bossThreadNumber = 0;
/**
* 负责处理事务的线程数，缺省是cpu的核心*2个数量，传0或负数则为缺省，大于100的数则为100
*/
private Integer workThreadNumber = 0;
/**
* 判断是否离线的时间间隔，缺省是90秒
*/
private Integer checkOfflineInterval = 90;
/**
* 最大一次性接收的报文长度，如果是图片之类的，这个值要设大一点，缺省是102400字节
*/
private Integer maxBytesInMessage = 102400;
/**
* 接收到的消息先存入队列，这个值是缺省的队列大小，如果队列满了，就无法接收新的数据，请确保快速处理队列里的数据
*/
private Integer maxMessageCount = 100000;
3. 实现 IMqttMessageListener 接口
最核心的处理是在 onMessage 方法里处理从客户端发送过来的数据。
/**
* 设备离线后触发，请不要在这个函数里添加耗时操作
*/
void offline(String uuid, ClientSession clientSession);
/**
* 接收到客户端返回的消息后触发，请不要在这个函数里添加耗时操作
*
* @param ip 客户端的ip地址
* @param channelId 通道id
* @param topic 接收到的消息TOPIC
* @param message 接收到的消息正文，已转换成字符串
* @param publisher 回调函数，如果需要给客户端发送数据，通过这个函数来处理
*/
void onMessage(String ip, String channelId, String topic, String message, Function<CustomMqttPublishMessage, Boolean> publisher);
/**
* 设备上线后触发，请不要在这个函数里添加耗时操作
*
* @param uuid 链接的设备的uuid
*/
void online(String uuid, ClientSession clientSession);
/**
* 某些特定设备的uuid并不是mqtt的clientId，需要做额外的处理
* 请不要在这个函数里添加耗时操作
*
* @param clientId mqtt client的唯一标识
* @return 根据clientId返回对应的uuid，绝大部分情况是相等
*/
String clientIdToUuid(String clientId);
4. 获取当前在线客户端列表和发送数据给客户端
内存里保存当前所有在线客户端对应的对象，ClientSession对象，包含了客户端 uuid、ip、回调函数等重要属性.
class ClientSession {
/**
* mqtt连接的用户名
*/
private String username;
/**
* mqtt连接的密码
*/
private byte[] password;
/**
* mqtt连接的ip
*/
private String ip;
/**
* 设备唯一标识
*/
private String uuid;
/**
* 设备对应的通信通道id
*/
private String channelId;
/**
* 给设备发送消息的回调函数
*/
private Function<CustomMqttPublishMessage, Boolean> publisher;
/**
* 设备最后刷新时间，也就是最后一次给paas发送数据的时间
*/
private Calendar refreshTime;
/**
* 一些额外的数据，用于扩展
*/
private Object data;
}
通常使用场景有2个：
- 获取所有在线客户端列表
- 根据客户端uuid获取对应对象的回调函数，通过回调函数来发送数据给客户端
//获取所有在线客户端列表
Map<String, ClientSession> clients = mqttserver.getClientSessionManager().getClientsMap();
//根据客户端uuid获取对应的对象
ClientSession client = clients.get("uuid123");
//构建一个要发送给客户端的数据结构
CustomMqttPublishMessage message = new CustomMqttPublishMessage(client.getChannelId(),"mytopic","mycontent");
//发送数据给客户端
client.getPublisher().apply(message);
三. 项目说明
1. mqttserver 子目录
这个目录是这个库的核心代码，只依赖了netty-all:4.1.86.Final
2. src 子目录
是测试mqttserver库的测试代码 里面依赖了一些其它库，需要用到 jdk 17
3. test 子目录
test目录里是用python模拟2000个mqtt客户端来同时链接这个mqttserver，用于功能测试和压力测试
