物联网 IOT设备数据采集平台
测试设备：有人DTU、温湿度设备传感器
一、采用的协议
TCP、MQTT server、MQTT client、HTTP
使用RabbitMQ实现MQTT 服务器，netty实现TCP，netty实现内嵌MQTT服务器，Vertx实现HTTP协议
二、实现过程
基础框架采用若依，时序数据库采用TDengine（存储高频数据，温湿度数据），使用JS的方式实现数据解析（NashornScriptEngine引擎实现解析）
HTTP协议
<!--vertx 模块-->
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-web</artifactId>
<version>3.5.2</version>
</dependency>
<!-- js 脚本解析-->
<dependency>
<groupId>org.openjdk.nashorn</groupId>
<artifactId>nashorn-core</artifactId>
<version>15.4</version>
</dependency>
创建HTTP 服务
public class HttpServer {
private Vertx vertx ;
public void startHttp(Integer port){
vertx = Vertx.vertx();
// 创建一个HttpServer
io.vertx.core.http.HttpServer server = vertx.createHttpServer();
// 创建路由对象
Router router = Router.router(vertx);
RouterDeal.dealRouter(router);
server.requestHandler(router::accept);
server.listen(port); // 监听8888端口
System.out.println("http启动成功，端口："+port);
}
public void close(){
vertx.close();
}
}监听路由，并且从http中获取参数（后续还可以进行接口的验签等操作 在TODO中实现）
public class RouterDeal {
public static void dealRouter(Router router){
//属性上报
router.route(HttpMethod.GET,"/:clientId/report/reportTopic").handler(rc -> {
final String[] clientId = {rc.request().getParam("clientId")};
//获取参数
MultiMap entries = rc.queryParams();
entries.forEach(entry -> {
clientId[0] = clientId[0] +" - " +entry.getKey()+":"+entry.getValue();
});
// TODO Handle the lookup of the product....
rc.response().end(clientId[0] );
});
//事件上报
router.route("/:clientId/event/eventTopic").handler(rc -> {
// TODO Handle the lookup of the product....
rc.response().end("clientId");
});
//设备上线
router.route("/:clientId/online/onlineTopic").handler(rc -> {
// TODO Handle the lookup of the product....
rc.response().end("clientId");
});
//设备离线
router.route("/:clientId/dead/deadTopic").handler(rc -> {
// TODO Handle the lookup of the product....
rc.response().end("clientId");
});
router.route("/index").handler(rc -> {
// TODO Handle the lookup of the product....
rc.response().end("clientId");
});
}
}MQTT 协议
连接其他MQTT服务端（EMQX、自建的MQTT服务器等）
自己作为客户端连接服务器，订阅所有的topic 实现所有设备的数据采集
package com.ruoyi.iot.server.mqtt.client;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
/**
* @program:
* @ClassName MqttConfig
* @description: mqtt 配置
* @author: 陈永强
**/
@Slf4j
public class MqttConfig {
private String username;
private String password;
private String hostUrl;
private String clientId;
public MqttConfig (String username,String password,String hostUrl,String clientId) {
this.username = username;
this.password = password;
this.hostUrl = hostUrl;
this.clientId = clientId;
}
/**
* 客户端对象
*/
private MqttClient client;
/**
* 在bean初始化后连接到服务器
*/
public void init() {
this.connect();
}
public void sendInit(String topic,String message ) throws MqttException {
this.connect();
this.publish(topic,message);
this.disConnect();
}
/**
* 断开连接
*/
public void disConnect() {
try {
this.client.disconnect();
this.client.close();
log.info("关闭服务");
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 客户端连接服务端
*/
public void connect() {
try {
// 创建MQTT客户端对象
client = new MqttClient(hostUrl, clientId, new MemoryPersistence());
// 连接设置
MqttConnectOptions options = new MqttConnectOptions();
// 是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos）,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
// 设置为true表示每次连接服务器都是以新的身份
options.setCleanSession(true);
// 设置连接用户名
options.setUserName(username);
// 设置连接密码
options.setPassword(password.toCharArray());
// 设置超时时间，单位为秒
options.setConnectionTimeout(100);
// 设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线
options.setKeepAliveInterval(20);
// 设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
options.setWill("willTopic", (clientId + "与服务器断开连接").getBytes(), 0, false);
// 设置回调
MqttCallBack mqttCallBack = new MqttCallBack(username,password,hostUrl,clientId);
client.setCallback(mqttCallBack);
// 连接
client.connect(options);
// 订阅主题 (接受此主题的消息)
this.subscribe("#", 2);
System.out.println("mqttClient连接成功");
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 发布消息
*
* @param topic
* @param message
*/
public boolean publish(String topic, String message) throws MqttException {
MqttMessage mqttMessage = new MqttMessage();
// 0：最多交付一次，可能丢失消息
// 1：至少交付一次，可能消息重复
// 2：只交付一次，既不丢失也不重复
mqttMessage.setQos(0);
// 是否保留最后一条消息
mqttMessage.setRetained(false);
byte[] bytes = message.getBytes();
// 消息内容
mqttMessage.setPayload(bytes);
// 主题的目的地，用于发布/订阅信息
MqttTopic mqttTopic = client.getTopic(topic);
// 提供一种机制来跟踪消息的传递进度
// 用于在以非阻塞方式（在后台运行）执行发布是跟踪消息的传递进度
MqttDeliveryToken token;
try {
// 将指定消息发布到主题，但不等待消息传递完成，返回的token可用于跟踪消息的传递状态
// 一旦此方法干净地返回，消息就已被客户端接受发布，当连接可用，将在后台完成消息传递。
token = mqttTopic.publish(mqttMessage);
token.waitForCompletion();
return true;
} catch (MqttException e) {
e.printStackTrace();
throw e;
}
}
/**
* 订阅主题
*/
public void subscribe(String topic, int qos) {
try {
client.setCallback(new MqttCallBack(username,password,hostUrl,clientId));
client.subscribe(topic, qos);
} catch (MqttException e) {
e.printStackTrace();
}
}
public static String hexToString(String hex) {
int decimal = Integer.parseInt(hex, 16);
return String.valueOf(decimal);
}
}
消息的接收和解析
package com.ruoyi.iot.server.mqtt.client;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.common.utils.spring.SpringUtils;
import com.ruoyi.iot.domain.IotDeviceAttribute;
import com.ruoyi.iot.domain.IotDeviceInfo;
import com.ruoyi.iot.service.IIotDeviceAttributeService;
import com.ruoyi.iot.service.IIotDeviceInfoService;
import com.ruoyi.iot.service.TaosService;
import com.ruoyi.iot.util.CommonParse;
import com.ruoyi.iot.util.MailUtil;
import com.ruoyi.iot.util.ParseUtil;
import io.netty.buffer.ByteBuf;
import io.netty.util.CharsetUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import java.nio.charset.StandardCharsets;
import java.util.List;
@Slf4j
@Data
public class MqttCallBack implements MqttCallback {
private String username;
private String password;
private String hostUrl;
private String clientId;
public MqttCallBack (String username,String password,String hostUrl,String clientId) {
this.username = username;
this.password = password;
this.hostUrl = hostUrl;
this.clientId = clientId;
}
/**
* 与服务器断开的回调
*/
@Override
public void connectionLost(Throwable cause) {
log.error("与服务器断开连接");
log.error("重连");
cause.printStackTrace();
MqttConfig bean = new MqttConfig(username,password,hostUrl,clientId+10);
bean.connect();
}
/**
* 消息发布成功的回调
*/
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
IMqttAsyncClient client = token.getClient();
log.info("【发送】【{}】##########发布消息成功！",client.getClientId());
}
/**
* 订阅消息接收 消息到达的回调
* */
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
//不解析离线和上线消息
log.info("mqtt消息");
if(!dealTopic(topic)) {
return;
}
//设备信息表
IIotDeviceInfoService deviceInfoService = SpringUtils.getBean(IIotDeviceInfoService.class);
//获取设备id
Long deviceId = Long.valueOf(topic.split("/")[1]);
//设备属性表
IIotDeviceAttributeService attributeService = SpringUtils.getBean(IIotDeviceAttributeService.class);
List<IotDeviceAttribute> attribute = attributeService.getByDeviceId(deviceId);
//获取设备id
if(ObjectUtil.isEmpty(deviceId)){
return;
}
IotDeviceInfo deviceInfo = deviceInfoService.getById(deviceId);
if(ObjectUtil.isEmpty(deviceInfo)){
return;
}
if(!deviceInfo.getSubTopic().equals(topic)){
return;
}
String msg = null;
if(!ObjectUtil.isEmpty(deviceInfo.getParseCode())){
msg = CommonParse.ParseHex(message, deviceInfo.getParseCode());
}else {
message.getPayload();
byte[] bytes = message.getPayload();
msg = new String(bytes, CharsetUtil.UTF_8);
}
//涛思数据库
TaosService taosService = SpringUtils.getBean(TaosService.class);
// MailUtil mail = SpringUtils.getBean(MailUtil.class);
// mail.sendAlarm(deviceId, JSONObject.parseObject(msg));
taosService.insertTables(deviceId,attribute,msg);
log.info("消息解析{}",new String(message.getPayload(), StandardCharsets.UTF_8));
}
//topic处理
public Boolean dealTopic(String topic){
//在线 online 下线 dead 上报 report
return topic.split("report").length > 1;
}
}
RabbitMQ工具实现设备连接状态
package com.ruoyi.iot.server.mqtt.client;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.iot.util.RabbitMQConnectionInfo;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
/**
* @author 陈永强
* @date 2024-12-03 15:03:27
*/
public class RabbitMQUtil {
public static List<RabbitMQConnectionInfo> getConnectInfo(String user, String pwd, String urls) {
try {
// 创建 URL 对象
URL url = new URL(urls);
HttpURLConnection connection = (HttpURLConnection) url.openConnection();
// 设置认证信息
String auth = user + ":" + pwd;
String encodedAuth = java.util.Base64.getEncoder().encodeToString(auth.getBytes("UTF-8"));
connection.setRequestProperty("Authorization", "Basic " + encodedAuth);
List<RabbitMQConnectionInfo> rabbitMQConnectionList = new ArrayList<>();
// 发送请求并获取响应
int responseCode = connection.getResponseCode();
if (responseCode == HttpURLConnection.HTTP_OK) {
BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
String inputLine;
StringBuilder response = new StringBuilder();
while ((inputLine = in.readLine()) != null) {
response.append(inputLine);
}
in.close();
String jsonString = response.toString().replace("\"client_properties\":{},","");
// 解析 JSON 响应
JSONArray array = JSONArray.parseArray(jsonString);
array.forEach(o->{
RabbitMQConnectionInfo rabbitMQConnection1 = JSONObject.parseObject(String.valueOf(o), RabbitMQConnectionInfo.class);
rabbitMQConnectionList.add(rabbitMQConnection1);
System.out.println("客户端id: " + rabbitMQConnection1.getClientProperties().getClientId());
});
System.out.println("当前连接数量: " + rabbitMQConnectionList.size());
} else {
System.out.println("无法获取连接信息，HTTP 响应码: " + responseCode);
}
// 关闭连接
connection.disconnect();
return rabbitMQConnectionList;
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
}
内嵌MQTT服务器
Netty 实现的MQTT 内嵌服务
服务端启动，消息接受，基础配置
package com.ruoyi.iot.server.mqtt.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import javax.annotation.PreDestroy;
@Slf4j
public class MqttServer {
private static NettyServerChannelInitializer mqttServerChannelInitializer;
private NioEventLoopGroup bossGroup;
private NioEventLoopGroup workerGroup;
private ChannelFuture future;
private int socketPort;
private String userName;
private String password;
public void start(Integer port,String userName,String password) {
this.userName = userName;
this.password = password;
socketPort = port;
mqttServerChannelInitializer = new NettyServerChannelInitializer(userName,password,socketPort);
//创建接收请求和处理请求的实例（默认线程数为 CPU 核心数乘以2也可自定义）
bossGroup = new NioEventLoopGroup(3);
workerGroup = new NioEventLoopGroup(6);
try {
//创建服务端启动辅助类(boostrap 用来为 Netty 程序的启动组装配置一些必须要组件，例如上面的创建的两个线程组)
ServerBootstrap socketBs = new ServerBootstrap();
//channel 方法用于指定服务器端监听套接字通道
//socket配置
socketBs.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
//ChannelOption.SO_BACKLOG对应的是tcp/ip协议listen函数中的backlog参数，
// 函数listen(int socketfd,int backlog)用来初始化服务端可连接队列，
// 服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接，
// 多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog参数指定了队列的大小
.option(ChannelOption.SO_BACKLOG, 1024)
//快速复用,防止服务端重启端口被占用的情况发生
.option(ChannelOption.SO_REUSEADDR, true)
.childHandler(mqttServerChannelInitializer)
//如果TCP_NODELAY没有设置为true,那么底层的TCP为了能减少交互次数,会将网络数据积累到一定的数量后,
// 服务器端才发送出去,会造成一定的延迟。在互联网应用中,通常希望服务是低延迟的,建议将TCP_NODELAY设置为true
.childOption(ChannelOption.TCP_NODELAY, true)
//默认的心跳间隔是7200s即2小时。Netty默认关闭该功能。
.childOption(ChannelOption.SO_KEEPALIVE, true);
future = socketBs.bind(socketPort).sync();
if (future.isSuccess()) {
log.info("[MQTT驱动服务端启动成功]");
future.channel().closeFuture().sync();
} else {
log.info("[~~~MQTT驱动服务端启动失败~~~]");
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
/**
* 这里可以手动关闭，同时在整个springboot应用停止的时候，这里也得以调用【@PreDestroy】
* 参考 https://www.cnblogs.com/CreatorKou/p/11606870.html
*/
public void shutdown() {
// 优雅关闭两个 EventLoopGroup 对象
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
log.info("[*MQTT服务端关闭成功]");
}
}
package com.ruoyi.iot.server.mqtt.server;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
/**
* @description: 配置管道 服务端初始化，客户端与服务器端连接一旦创建，这个类中方法就会被回调，设置出站编码器和入站解码器
* @create: 2020-04-03 14:14
**/
public class NettyServerChannelInitializer extends ChannelInitializer<SocketChannel> {
private String userName;
private String password;
private Integer port;
private static ServerMqttHandler serverMqttHandler ;
public NettyServerChannelInitializer(String userName, String password,Integer port) {
this.password = password;
this.userName = userName;
this.port = port;
}
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
serverMqttHandler = new ServerMqttHandler(userName,password,port);
pipeline.addLast("decoder", new MqttDecoder(1024 * 8));
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
pipeline.addLast(serverMqttHandler);
}
}
package com.ruoyi.iot.server.mqtt.server;
import cn.hutool.core.collection.CollUtil;
import com.ruoyi.iot.util.OfflineUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
@Slf4j
public class ServerMqttHandler extends SimpleChannelInboundHandler<MqttMessage> {
private String userName;
private String password;
private Integer port;
//客户端
public static final ConcurrentHashMap<String, ChannelHandlerContext> clientMap = new ConcurrentHashMap<String, ChannelHandlerContext>();
// @Value("${driver.mqtt.address_list}") //白名单
// private List<String> addressList;
private static MqttMsgBack mqttMsgBack ;
public ServerMqttHandler(String userName, String password,Integer port) {
this.password = password;
this.userName = userName;
this.port = port;
mqttMsgBack = new MqttMsgBack(userName,password,port);
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
//判断连接是否合法
// if (addressList.contains(address.getHostString())) {
// clientMap.put(ctx.channel().id().toString(), ctx);
// super.handlerAdded(ctx);
// } else {
// ctx.close();
// }
//允许所有连接
clientMap.put(ctx.channel().id().toString()+":"+port, ctx);
super.handlerAdded(ctx);
}
/**
* 功能描述: 客户端终止连接服务器会触发此函数
*
* @param ctx
* @return void
* @author zhouwenjie
* @date 2020/4/3 16:47
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
//离线
System.out.println("设备离线");
AtomicReference<String> deviceIdAtomic = new AtomicReference<>();
ConcurrentHashMap<String, String> clients = MqttMsgBack.getClientMap();
clients.keySet().forEach(r->{
if(clients.get(r).equals(ctx.channel().id().toString())){
deviceIdAtomic.set(r);
}
});
String deviceId = deviceIdAtomic.get();
OfflineUtil.sendMsg(deviceId);
System.out.println(deviceId+": 设备离线");
delSubCache(ctx);
super.channelInactive(ctx);
}
/**
* 功能描述: 有客户端发消息会触发此函数
*
* @param ctx
* @param mqttMessage
* @return void
* @author zhouwenjie
* @date 2020/4/3 16:48
*/
@Override
public void channelRead0(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
if (null != mqttMessage) {
log.info("接收mqtt消息{} 端口：{}" , "mqttMessage",port);
MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();
switch (mqttFixedHeader.messageType()) {
// ----------------------接收消息端（服务端）可能会触发的事件----------------------------------------------------------------
case CONNECT:
// 在一个网络连接上，客户端只能发送一次CONNECT报文。服务端必须将客户端发送的第二个CONNECT报文当作协议违规处理并断开客户端的连接
// 建议connect消息单独处理，用来对客户端进行认证管理等 这里直接返回一个CONNACK消息
mqttMsgBack.connectionAck(ctx, mqttMessage);
break;
case PUBLISH:
// 收到消息，返回确认，PUBACK报文是对QoS 1等级的PUBLISH报文的响应,PUBREC报文是对PUBLISH报文的响应
mqttMsgBack.publishAck(ctx, mqttMessage,port);
break;
case PUBREL:
// 释放消息，PUBREL报文是对QoS 2等级的PUBREC报文的响应,此时我们应该回应一个PUBCOMP报文
mqttMsgBack.publishComp(ctx, mqttMessage);
break;
case SUBSCRIBE:
// 客户端订阅主题
// 客户端向服务端发送SUBSCRIBE报文用于创建一个或多个订阅，每个订阅注册客户端关心的一个或多个主题。
// 为了将应用消息转发给与那些订阅匹配的主题，服务端发送PUBLISH报文给客户端。
// SUBSCRIBE报文也（为每个订阅）指定了最大的QoS等级，服务端根据这个发送应用消息给客户端
mqttMsgBack.subscribeAck(ctx, mqttMessage);
break;
case UNSUBSCRIBE:
// 客户端取消订阅
// 客户端发送UNSUBSCRIBE报文给服务端，用于取消订阅主题
mqttMsgBack.unsubscribeAck(ctx, mqttMessage);
break;
case PINGREQ:
// 客户端发起心跳
mqttMsgBack.pingResp(ctx, mqttMessage);
break;
case DISCONNECT:
// 客户端主动断开连接
// DISCONNECT报文是客户端发给服务端的最后一个控制报文， 服务端必须验证所有的保留位都被设置为0
break;
// ----------------------服务端作为发送消息端可能会接收的事件----------------------------------------------------------------
case PUBACK:
case PUBREC:
//QoS 2级别,响应一个PUBREL报文消息，PUBACK、PUBREC这俩都是ack消息
//PUBACK报文是对QoS 1等级的PUBLISH报文的响应，如果一段时间没有收到客户端ack，服务端会重新发送消息
mqttMsgBack.receivePubAck(ctx, mqttMessage);
break;
case PUBCOMP:
//收到qos2级别接收端最后一次发送过来的确认消息
mqttMsgBack.receivePubcomp(ctx, mqttMessage);
break;
default:
break;
}
}
}
/**
* 功能描述: 心跳检测
*
* @param ctx 这里的作用主要是解决断网，弱网的情况发生
* @param evt
* @return void
* @author zhouwenjie
* @date 2020/4/3 17:02
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
String socketString = ctx.channel().remoteAddress().toString();
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state() == IdleState.READER_IDLE) {
log.info("Client: " + socketString + " READER_IDLE 读超时");
delSubCache(ctx);
ctx.disconnect();
}
}
}
/**
* 功能描述:
*
* @param ctx
* @param cause
* @return void
* @author 发生异常会触发此函数
* @date 2020/4/3 16:49
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
}
private void delSubCache(ChannelHandlerContext ctx){
String id = ctx.channel().id().toString();
mqttMsgBack.removeClientMap(id);
clientMap.remove(id+":"+port);
//删除订阅主题
Set<String> topicSet = MqttMsgBack.ctMap.get(id);
if (CollUtil.isNotEmpty(topicSet)) {
ConcurrentHashMap<String, HashSet<String>> subMap = MqttMsgBack.subMap;
ConcurrentHashMap<String, MqttQoS> qoSMap = MqttMsgBack.qoSMap;
for (String topic : topicSet) {
if (subMap != null) {
HashSet<String> ids = subMap.get(topic);
if (CollUtil.isNotEmpty(ids)) {
ids.remove(id);
if (CollUtil.isEmpty(ids)) {
subMap.remove(topic);
}
}
}
if (qoSMap != null) {
qoSMap.remove(topic + "-" + id);
}
}
}
MqttMsgBack.ctMap.remove(id);
}
public static Set<String> getClientId(){
ConcurrentHashMap<String, String> client = MqttMsgBack.getClientMap();
return new HashSet<>(client.keySet());
}
}