物联网 IOT设备数据采集平台 二
内嵌MQTT 服务器实现类
package com.ruoyi.iot.server.mqtt.server;
import cn.hutool.core.collection.CollUtil;
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.mapper.IotDeviceAttributeMapper;
import com.ruoyi.iot.mapper.IotDeviceInfoMapper;
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.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import static io.netty.handler.codec.mqtt.MqttMessageType.PUBACK;
import static io.netty.handler.codec.mqtt.MqttMessageType.PUBREC;
import static io.netty.handler.codec.mqtt.MqttQoS.*;
/**
* @description: 对接收到的消息进行业务处理
* CONNECT 1 C->S 客户端请求与服务端建立连接 （服务端接收）
* CONNACK 2 S->C 服务端确认连接建立（客户端接收）
* PUBLISH 3 CóS 发布消息 （服务端接收【QoS(服务质量等级) 0级别，最多分发一次】）-->生产者只会发送一次消息，不关心消息是否被代理服务端或消费者收到
* PUBACK 4 CóS 收到发布消息确认（客户端接收【QoS 1级别，至少分发一次】） -->保证消息发送到服务端（也就是代理服务器broker），如果没收到或一定时间没收到服务端的ack，就会重发消息
* PUBREC 5 CóS 收到发布消息（客户端接收【QoS 2级别】）|
* PUBREL 6 CóS 释放发布消息（服务端接收【QoS 2级别】）|只分发一次消息，且保证到达 -->这三步保证消息有且仅有一次传递给消费者
* PUBCOMP 7 CóS 完成发布消息（客户端接收【QoS 2级别】）|
* SUBSCRIBE 8 C->S 订阅请求（服务端接收）
* SUBACK 9 S->C 订阅确认（客户端接收）
* UNSUBSCRIBE 10 C->S 取消订阅（服务端接收）
* UNSUBACK 11 S->C 取消订阅确认（客户端接收）
* PINGREQ 12 C->S 客户端发送PING(连接保活)命令（服务端接收）
* PINGRESP 13 S->C PING命令回复（客户端接收）
* DISCONNECT 14 C->S 断开连接 （服务端接收）
* <p>
* 注意：在我们发送消息的时候，一定要确认好等级，回复确认的消息统一设置为qos=0;
* 比如，我们需要发送最高等级的消息就将qos设置为2，当我们接收方收到这个等级的消息的时候判断一下等级，设置好消息类型，然后qos因为是回复消息，
* 所以全部设置成0，而当发送端收到消息之后，比如PUBREC，那么我们还将发送（注意不是回复）PUBREL给接收端，希望对方回复确认一下，所以qos设置为1。
* 综上可知：发送端没有回复确认消息之说，只有发送消息，接收端没有发送消息之说，只有回复确认消息，搞清楚这个概念，在设置参数的时候就明了。
**/
@Slf4j
@Data
public class MqttMsgBack {
private String userName;
private String password;
private Integer port;
public MqttMsgBack(String userName,String password,Integer port){
this.password = password;
this.userName = userName;
this.port = port;
}
private final long waitTime = 100000L;
/**
* 功能描述:存放主题和其订阅的客户端集合
*/
public static final ConcurrentHashMap<String, HashSet<String>> subMap = new ConcurrentHashMap<String, HashSet<String>>();
/**
* 功能描述:存放设备id
*/
public static final ConcurrentHashMap<String, String> clientMap = new ConcurrentHashMap<String, String>();
/**
* 功能描述:存放订阅是的服务质量等级，只有发送小于或等于这个服务质量的消息给订阅者
*/
public static final ConcurrentHashMap<String, MqttQoS> qoSMap = new ConcurrentHashMap<String, MqttQoS>();
/**
* 功能描述:存放客户端和其所订阅的主题集合，用来在客户端断开的时候删除订阅中的客户端
*/
public static final ConcurrentHashMap<String, Set<String>> ctMap = new ConcurrentHashMap<String, Set<String>>();
/**
* 功能描述:存放需要缓存的消息，一边发送给新订阅的客户端
*/
public static final ConcurrentHashMap<String, MqttPublishMessage> cacheRetainedMessages = new ConcurrentHashMap<String, MqttPublishMessage>();
/**
* 功能描述:缓存需要重复发送的消息，以便在收到ack的时候将消息内存释放掉
*/
public static final ConcurrentHashMap<String, ByteBuf> cacheRepeatMessages = new ConcurrentHashMap<String, ByteBuf>();
/**
* 确认连接请求
*
* @param ctx
* @param mqttMessage
*/
public void connectionAck(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
MqttConnectMessage mqttConnectMessage;
try {
mqttConnectMessage = (MqttConnectMessage) mqttMessage;
//获取连接者的ClientId
String clientIdentifier = mqttConnectMessage.payload().clientIdentifier();
//查询用户名密码是否正确
String userNameNow = mqttConnectMessage.payload().userName();
String passwordNow = mqttConnectMessage.payload().password();
if (userName.equals(userNameNow) && password.equals(passwordNow)) {
MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
MqttConnectVariableHeader mqttConnectVariableHeaderInfo = mqttConnectMessage.variableHeader();
//构建返回报文， 可变报头
MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, mqttConnectVariableHeaderInfo.isCleanSession());
//构建返回报文， 固定报头 至多一次(至少—次,只有一次)
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK, mqttFixedHeaderInfo.isDup(), AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
//构建连接回复消息体
MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);
ctx.writeAndFlush(connAck);
//设置节点名
InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
clientMap.put(clientIdentifier,ctx.channel().id().toString());
log.info("终端登录成功,ID号:{},IP信息:{},终端号:{}", clientIdentifier, address.getHostString(), address.getPort());
} else {
//如果用户名密码错误则提示对方
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, AT_MOST_ONCE, false, 0x02);
MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false);
MqttConnAckMessage mqttConnAckMessage = new MqttConnAckMessage(fixedHeader, variableHeader);
ctx.writeAndFlush(mqttConnAckMessage);
ctx.close();
log.error("连接失败：" + MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD.name());
}
} catch (ClassCastException e) {
//转换失败，对方发送的协议版本不兼容
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, AT_MOST_ONCE, false, 0x02);
MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false);
MqttConnAckMessage mqttConnAckMessage = new MqttConnAckMessage(fixedHeader, variableHeader);
ctx.writeAndFlush(mqttConnAckMessage);
ctx.close();
e.printStackTrace();
log.error("连接失败：" + MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION.name());
}
}
/**
* 根据qos发布确认
* 1：表示发送的消息需要一直持久保存（不受服务器重启影响），不但要发送给当前的订阅者，并且以后新来的订阅了此Topic name的订阅者会马上得到推送。
* 0：仅仅为当前订阅者推送此消息。
* 假如服务器收到一个空消息体(zero-length payload)、RETAIN = 1、那么就代表需要将这个缓存消息删除掉，不再继续推送给新订阅者，前提是一定保证空消息体。
* 两个条件需要对应上才能删除，消息体为空、主题名称对应.
* 注意：对于每个主题，只能保留一条消息。当发布一个带有RETAIN标志的新消息时，它将替换上一条保留的消息，因此在同一主题下只会存在一条保留消息。
*
* @param ctx
* @param mqttMessage 拷贝的数据需要释放，其他的消息，比如连接消息，不用转发，会在handler中自动释放
*/
public void publishAck(ChannelHandlerContext ctx, MqttMessage mqttMessage,Integer port) {
MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
MqttFixedHeader mqttFixedHeaderInfo = mqttPublishMessage.fixedHeader();
MqttQoS qos = mqttFixedHeaderInfo.qosLevel();
//得到主题
MqttPublishVariableHeader variableHeader = mqttPublishMessage.variableHeader();
String topicName = variableHeader.topicName();
//保存上报消息
setSubAttribute(topicName,mqttMessage);
//将消息发送给订阅的客户端
ByteBuf byteBuf = mqttPublishMessage.payload();
// byte[] bytes = new byte[byteBuf.readableBytes()];
// byteBuf.readBytes(bytes);
// byteBuf.readBytes(bytes); 重复读取存在问题，不可重复读
// String s = new String(bytes, CharsetUtil.UTF_8);
// log.info("消息内容：{}",s);
HashSet<String> set = subMap.get(topicName+":"+port);
// # 代表监听全部
if(ObjectUtil.isEmpty(set)){
set = new HashSet<>();
}
if(!ObjectUtil.isEmpty(subMap.get("#:"+port))){
set.addAll(subMap.get("#:"+port));
}
if (set != null) {
for (String channelId : set) {
ChannelHandlerContext context = ServerMqttHandler.clientMap.get(channelId+":"+port);
if (context != null && context.channel().isActive()) {
MqttQoS cacheQos = qoSMap.get(topicName + "-" + channelId+":"+port);
if(ObjectUtil.isEmpty(cacheQos)){
cacheQos = qoSMap.get("#"+ "-" + channelId+":"+port);
}
if (cacheQos != null && qos.value() <= cacheQos.value()) {
// retainedDuplicate()增加引用计数器，不至于后续操作byteBuf出现错误，引用计数器为0的情况，这里会清除retainedDuplicate的操作有：
// SimpleChannelInboundHandler处理器、编码器 MqttEncoder、和最后确认的时候释放，所以每次操作消息之前，先进行一次retainedDuplicate
byteBuf.retainedDuplicate();
context.writeAndFlush(mqttPublishMessage);
if (qos == AT_LEAST_ONCE || qos == EXACTLY_ONCE) {
//只发送服务质量等级小于等于订阅时客户端指定的服务质量等级
// 创建一个新的缓冲区
byteBuf.retainedDuplicate();
//防止内存溢出，最后在消息被ack或者客户端断开掉线的时候，拿到并进行释放
cacheRepeatMessages.put(channelId, byteBuf);
cachePublishMsg(qos, byteBuf, variableHeader, mqttFixedHeaderInfo, context);
}
}else if(cacheQos != null){
byteBuf.retainedDuplicate();
context.writeAndFlush(mqttPublishMessage);
if (qos == AT_LEAST_ONCE || qos == EXACTLY_ONCE) {
//只发送服务质量等级小于等于订阅时客户端指定的服务质量等级
// 创建一个新的缓冲区
byteBuf.retainedDuplicate();
//防止内存溢出，最后在消息被ack或者客户端断开掉线的时候，拿到并进行释放
cacheRepeatMessages.put(channelId, byteBuf);
cachePublishMsg(qos, byteBuf, variableHeader, mqttFixedHeaderInfo, context);
}
}
} else {
if (context != null) {
InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
log.error(address.getHostString() + "转发订阅消息提醒：客户端连接异常~");
}
//防止客户端频繁上下线导致id变化，带来不必要的空指针
ServerMqttHandler.clientMap.remove(channelId);
//删除订阅主题
Set<String> topicSet = ctMap.get(channelId);
if (CollUtil.isNotEmpty(topicSet)) {
for (String topic : topicSet) {
if (subMap != null) {
HashSet<String> ids = subMap.get(topic+":"+port);
if (CollUtil.isNotEmpty(ids)) {
ids.remove(channelId);
if (CollUtil.isEmpty(ids)) {
subMap.remove(topic);
}
}
}
if (qoSMap != null) {
qoSMap.remove(topic + "-" + channelId+":"+port);
}
}
}
ctMap.remove(channelId);
}
}
}
// 缓存消息给后订阅的客户端
boolean retain = mqttFixedHeaderInfo.isRetain();
if (retain) {
if (byteBuf.readableBytes() > 0) {
byteBuf.retainedDuplicate();
cacheRetainedMessages.put(topicName, mqttPublishMessage);
} else {
MqttPublishMessage message = cacheRetainedMessages.get(topicName);
if (message != null) {
cacheRetainedMessages.remove(topicName);
// 这里需要手动删除ByteBuf缓存,因为一直会有一份缓存在内存中备用
boolean release = ReferenceCountUtil.release(message);
log.info("缓存消息给后订阅的客户端释放成功失败：{}", release);
}
}
}
//返回消息给发送端
switch (qos) {
//至多一次
case AT_MOST_ONCE:
break;
//至少一次
case AT_LEAST_ONCE:
//构建返回报文， 可变报头
MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
//构建返回报文， 固定报头
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(PUBACK, mqttFixedHeaderInfo.isDup(), AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
//构建PUBACK消息体
MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
ctx.writeAndFlush(pubAck);
break;
//刚好一次
case EXACTLY_ONCE:
//构建返回报文，固定报头
MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(PUBREC, false, AT_MOST_ONCE, false, 0x02);
//构建返回报文，可变报头
MqttPubReplyMessageVariableHeader mqttPubReplyMessageVariableHeader = new MqttPubReplyMessageVariableHeader(mqttPublishMessage.variableHeader().packetId(), MqttPubReplyMessageVariableHeader.REASON_CODE_OK, MqttProperties.NO_PROPERTIES);
MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2, mqttPubReplyMessageVariableHeader);
ctx.writeAndFlush(mqttMessageBack);
break;
default:
break;
}
}
private void cachePublishMsg(MqttQoS qos, ByteBuf byteBuf, MqttPublishVariableHeader variableHeader, MqttFixedHeader mqttFixedHeaderInfo, ChannelHandlerContext context) {
//缓存一份消息，规定时间内没有收到ack，用作重发，重发时将isDup设置为true,代表重复消息
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, true, qos, false, mqttFixedHeaderInfo.remainingLength());
MqttPublishMessage cachePubMessage = new MqttPublishMessage(fixedHeader, variableHeader, byteBuf);
ScheduledFuture<?> scheduledFuture = TimerData.scheduledThreadPoolExecutor.scheduleAtFixedRate(new MonitorMsgTime(variableHeader.packetId(), cachePubMessage, context), waitTime, waitTime, TimeUnit.MILLISECONDS);
TimerData.scheduledFutureMap.put(variableHeader.packetId(), scheduledFuture);
}
/**
* 发布完成 qos2
*
* @param ctx
* @param mqttMessage
*/
public void publishComp(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
//构建返回报文， 固定报头
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, AT_MOST_ONCE, false, 0x02);
//构建返回报文， 可变报头
MqttPubReplyMessageVariableHeader mqttPubReplyMessageVariableHeader = new MqttPubReplyMessageVariableHeader(messageIdVariableHeader.messageId(), MqttPubReplyMessageVariableHeader.REASON_CODE_OK, MqttProperties.NO_PROPERTIES);
MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack, mqttPubReplyMessageVariableHeader);
ctx.writeAndFlush(mqttMessageBack);
}
/**
* 订阅确认
* 订阅和取消订阅没有qos2级别，默认就是1级别
* 需要存储订阅主题和客户端、客户端和订阅主题、验证防止重复订阅、发送缓存消息
*
* @param ctx
* @param mqttMessage
*/
public void subscribeAck(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
//构建返回报文， 可变报头
MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
MqttSubscribePayload subscribePayload = mqttSubscribeMessage.payload();
List<MqttTopicSubscription> mqttTopicSubscriptions = subscribePayload.topicSubscriptions();
List<Integer> grantedQoSLevels = new ArrayList<Integer>();
String id = ctx.channel().id().toString();
//存储客户端订阅的主题集合，断开或者异常连接时，删除订阅ctMap和subMap里的值
Set<String> topicSet = ctMap.get(id);
if (topicSet == null) {
topicSet = new HashSet<String>();
}
for (MqttTopicSubscription subscription : mqttTopicSubscriptions) {
String topicName = subscription.topicName();
HashSet<String> contexts = subMap.get(topicName+":"+port);
if (contexts == null) {
contexts = new HashSet<String>();
}
//先判断主题是否已经订阅过了，防止重复订阅
boolean isSub = contexts.contains(topicName+":"+port);
if (!isSub) {
MqttQoS qos = subscription.option().qos();
//存储主题被订阅的客户端集合
contexts.add(id);
qoSMap.put(topicName + "-" + id+":"+port, qos);
subMap.put(topicName+":"+port, contexts);
//存储客户端订阅的主题集合
topicSet.add(topicName);
}
//存储客户端订阅的主题集合
int value = subscription.qualityOfService().value();
grantedQoSLevels.add(value);
}
// 构建返回报文 有效负载
MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSLevels);
// 构建返回报文 固定报头
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, AT_MOST_ONCE, false, 0);
// 构建返回报文 订阅确认
MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack, variableHeaderBack, payloadBack);
ctx.writeAndFlush(subAck);
for (String topic : topicSet) {
MqttQoS cacheQos = qoSMap.get(topic + "-" + id+":"+port);
//查看订阅的主题是否需要需要发送消息
MqttPublishMessage mqttMsg = cacheRetainedMessages.get(topic);
if (mqttMsg != null) {
MqttFixedHeader mqttFixedHeaderInfo = mqttMsg.fixedHeader();
MqttQoS qos = mqttFixedHeaderInfo.qosLevel();
if (cacheQos != null && qos.value() <= cacheQos.value()) {
if (mqttMsg != null) {
MqttPublishVariableHeader variableHeader = mqttMsg.variableHeader();
ByteBuf payload = mqttMsg.payload();
//引用计数器增加
payload.retainedDuplicate();
ctx.writeAndFlush(mqttMsg);
// 开启消息重发机制
if (qos == AT_LEAST_ONCE || qos == EXACTLY_ONCE) {
//引用计数器增加
payload.retainedDuplicate();
cacheRepeatMessages.put(id, payload);
cachePublishMsg(qos, payload, variableHeader, mqttFixedHeaderInfo, ctx);
}
}
}
}
}
}
/**
* 取消订阅确认
*
* @param ctx
* @param mqttMessage
*/
public void unsubscribeAck(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
MqttUnsubscribeMessage mqttUnsubscribeMessage = (MqttUnsubscribeMessage) mqttMessage;
MqttMessageIdVariableHeader messageIdVariableHeader = mqttUnsubscribeMessage.variableHeader();
// 构建返回报文 可变报头
MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
// 构建返回报文 固定报头
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_MOST_ONCE, false, 0x02);
// 构建返回报文 取消订阅确认
MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack, variableHeaderBack);
log.info("取消订阅回复:{}", unSubAck);
//删除本地订阅客户端
String id = ctx.channel().id().toString();
List<String> topics = mqttUnsubscribeMessage.payload().topics();
Set<String> topicSet = ctMap.get(id);
for (String topic : topics) {
if (subMap != null) {
HashSet<String> ids = subMap.get(topic+":"+port);
if (CollUtil.isNotEmpty(ids)) {
ids.remove(id);
if (CollUtil.isEmpty(ids)) {
subMap.remove(topic);
}
}
}
if (qoSMap != null) {
qoSMap.remove(topic + "-" + id+":"+port);
}
if (CollUtil.isNotEmpty(topicSet)) {
topicSet.remove(topic);
}
}
ctx.writeAndFlush(unSubAck);
}
/**
* 心跳响应
*
* @param ctx
* @param mqttMessage
*/
public void pingResp(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, AT_MOST_ONCE, false, 0);
MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
log.info("心跳回复:{}", mqttMessageBack.toString());
ctx.writeAndFlush(mqttMessageBack);
}
/**
* ------------------------------------------------------服务端作为发送消息端可能会接收的事件----------------------------------------------------------------
* <p>
* 收到接收方消息确认，qos>1的情况，应该删除消息缓存(缓存消息保存到线程定时中了，循环发送取消，消息缓存也没有了)，取消消息重发机制
*
* @param ctx
* @param mqttMessage
*/
public void receivePubAck(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
MqttFixedHeader fixedHeader = mqttMessage.fixedHeader();
MqttMessageType messageType = fixedHeader.messageType();
if (messageType == PUBACK) {
MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
int messageId = variableHeader.messageId();
//等级为1的情况，直接删除原始消息，取消消息重发机制
ScheduledFuture<?> scheduledFuture = TimerData.scheduledFutureMap.remove(messageId);
if (scheduledFuture != null) {
scheduledFuture.cancel(true);
}
//移除消息记录
ByteBuf byteBuf = cacheRepeatMessages.remove(ctx.channel().id().toString());
if (byteBuf != null) {
// 释放内存
byteBuf.release();
}
}
if (messageType == PUBREC) {
//等级为2的情况，收到PUBREC报文消息，先停止消息重发机制，再响应一个PUBREL报文并且构建消息重发机制
MqttPubReplyMessageVariableHeader variableHeader = (MqttPubReplyMessageVariableHeader) mqttMessage.variableHeader();
int messageId = variableHeader.messageId();
//构建返回报文，固定报头
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0);
//构建返回报文，可变报头
MqttPubReplyMessageVariableHeader mqttPubReplyMessageVariableHeader = new MqttPubReplyMessageVariableHeader(messageId, MqttPubReplyMessageVariableHeader.REASON_CODE_OK, MqttProperties.NO_PROPERTIES);
MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack, mqttPubReplyMessageVariableHeader);
//删除初始消息重发机制
ScheduledFuture<?> scheduledFuture = TimerData.scheduledFutureMap.remove(messageId);
if (scheduledFuture != null) {
scheduledFuture.cancel(true);
}
//释放消息缓存
ByteBuf byteBuf = cacheRepeatMessages.remove(ctx.channel().id().toString());
if (byteBuf != null) {
byteBuf.release();
}
ctx.writeAndFlush(mqttMessageBack);
//重发机制要放在最下方，否则，一旦出错，会多次出发此机制
cachePubrelMsg(messageId, ctx);
}
}
private void cachePubrelMsg(int messageId, ChannelHandlerContext context) {
//缓存一份消息，规定时间内没有收到ack，用作重发，重发时将isDup设置为true,代表重复消息
//构建返回报文，固定报头
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBREL, true, AT_LEAST_ONCE, false, 0);
//构建返回报文，可变报头
MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(messageId);
MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
ScheduledFuture<?> scheduledFuture = TimerData.scheduledThreadPoolExecutor.scheduleAtFixedRate(new MonitorMsgTime(messageId, mqttMessageBack, context), waitTime, waitTime, TimeUnit.MILLISECONDS);
TimerData.scheduledFutureMap.put(messageId, scheduledFuture);
}
/**
* 功能描述: 接收到最后一次确认，取消上次PUBREL的消息重发机制
*
* @param ctx
* @param mqttMessage
* @return void
*/
public void receivePubcomp(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
MqttPubReplyMessageVariableHeader variableHeader = (MqttPubReplyMessageVariableHeader) mqttMessage.variableHeader();
int messageId = variableHeader.messageId();
ScheduledFuture<?> scheduledFuture = TimerData.scheduledFutureMap.remove(messageId);
if (scheduledFuture != null) {
scheduledFuture.cancel(true);
}
ByteBuf byteBuf = cacheRepeatMessages.remove(ctx.channel().id().toString());
if (byteBuf != null) {
byteBuf.release();
}
}
public static ConcurrentHashMap<String,String> getClientMap(){
return clientMap;
}
public void removeClientMap(String ctxId){
for (String key:clientMap.keySet()){
if(clientMap.get(key).equals(ctxId)){
clientMap.remove(key);
}
}
}
//保存设备上报的消息
public void setSubAttribute(String topic,MqttMessage mqttMessage){
// MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;
// //获取连接者的ClientId
// String clientIdentifier = mqttConnectMessage.payload().clientIdentifier();
// if(!clientIdentifier.equals(topic.split("/")[1])){
// return;
// }
MqttPublishMessage mqttPublishMessage = (MqttPublishMessage)mqttMessage;
//设备信息表
IIotDeviceInfoService deviceInfoService = SpringUtils.getBean(IIotDeviceInfoService.class);
//获取设备id
Long deviceId = Long.valueOf(topic.split("/")[1]);
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())){
ByteBuf byteBuf = mqttPublishMessage.payload();
byte[] bytes = new byte[byteBuf.readableBytes()];
// byteBuf.readBytes(bytes); 重复读取存在问题，不可重复读
ByteBuf byteBuf1 = byteBuf.duplicate();
byteBuf1.readBytes(bytes);
msg = CommonParse.ParseHex(bytes, deviceInfo.getParseCode());
}else {
ByteBuf byteBuf = mqttPublishMessage.payload();
byte[] bytes = new byte[byteBuf.readableBytes()];
// byteBuf.readBytes(bytes); 重复读取存在问题，不可重复读
ByteBuf byteBuf1 = byteBuf.duplicate();
byteBuf1.readBytes(bytes);
msg = new String(bytes, CharsetUtil.UTF_8);
}
//设备属性表
IIotDeviceAttributeService attributeService = SpringUtils.getBean(IIotDeviceAttributeService.class);
List<IotDeviceAttribute> attribute = attributeService.getByDeviceId(deviceId);
//涛思数据库
TaosService taosService = SpringUtils.getBean(TaosService.class);
taosService.insertTables(deviceId,attribute,msg);
//预警
// MailUtil mail = SpringUtils.getBean(MailUtil.class);
// mail.sendAlarm(deviceId, JSONObject.parseObject(msg));
}
}
实现内嵌的MQTT服务器
TCP 协议
使用Netty 实现TCP协议
package com.ruoyi.iot.server.tcp;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.ruoyi.iot.util.ParseUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
@Slf4j
public class BootNettyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter {
private static BootNettyChannelInboundHandlerAdapter nettyServerHandler;
public void init() {
nettyServerHandler = this;
}
/**
* 从客户端收到新的数据时，这个方法会在收到消息时被调用
*
* @param ctx
* @param msg
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception, IOException {
try{
if (msg == null) {
log.error("======加载客户端报文为空======【" + ctx.channel().id() + "】" + " :" + msg);
}
//解析数据
String clientId = getClientId(msg);
if(ObjectUtils.isEmpty(clientId)){
System.out.println("============报文数据==============");
System.out.println(msg);
System.out.println("================================");
ctx.write(ctx.channel().id()+"服务端已成功接收数据！");
return;
}
//获取解析数据后用户绑定
UserChannelMap.put(clientId,ctx.channel());
log.info("======加载客户端报文======【" + ctx.channel().id() + "】" + msg);
// 解析数据
getGateWayData(msg);
//回应客户端
ctx.write(ctx.channel().id()+"服务端已成功接收数据！");
}catch (Exception e){
e.printStackTrace();
}
}
private void getGateWayData(Object msg){
String s = ParseUtil.parseParams(msg);
// log.info("消息解析{}",s);
// JSONObject jsonObject = JSONObject.parseObject(String.valueOf(msg));
log.info("==========解析网关广播数据："+s);
}
private String getClientId(Object msg){
try{
JSONObject jsonObject = JSONObject.parseObject(String.valueOf(msg));
return String.valueOf(jsonObject.get("clientId"));
}catch (Exception e){
return null;
}
}
public static String hexToString(String hex) {
int decimal = Integer.parseInt(hex, 16);
return String.valueOf(decimal);
}
/**
* 从客户端收到新的数据、读取完成时调用
*
* @param ctx
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {
log.info("channelReadComplete");
ctx.flush();
}
/**
* 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
*
* @param ctx
* @param cause
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws IOException {
log.info("exceptionCaught");
cause.printStackTrace();
ctx.close();//抛出异常，断开与客户端的连接
}
/**
* 客户端与服务端第一次建立连接时 执行
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelActive(ctx);
ctx.channel().read();
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
//此处不能使用ctx.close()，否则客户端始终无法与服务端建立连接
log.info("channelActive:"+clientIp+ctx.name());
}
/**
* 客户端与服务端 断连时 执行
*
* @param ctx
* @throws Exception
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelInactive(ctx);
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
ctx.close(); //断开连接时，必须关闭，否则造成资源浪费，并发量很大情况下可能造成宕机
log.info("channelInactive:"+clientIp);
}
/**
* 服务端当read超时, 会调用这个方法
*
* @param ctx
* @param evt
* @throws Exception
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException {
super.userEventTriggered(ctx, evt);
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
//移除客户端
UserChannelMap.removechannel(String.valueOf(ctx.channel().id().asLongText()));
//超时时断开连接
ctx.close();
log.info("userEventTriggered:"+clientIp+" - "+ "通道id:"+String.valueOf(ctx.channel().id()));
}
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception{
log.info("channelRegistered:{}","注册设备:"+ctx.channel().id());
}
@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception{
//当一个客户端断开连接或一个服务器完成其服务并关闭连接时
//移除设备
UserChannelMap.removechannel(String.valueOf(ctx.channel().id().asLongText()));
//超时时断开连接
ctx.close();
log.info("channelUnregistered"+ctx.channel().id().asLongText());
}
/**
* 流量控制是一种拥塞控制机制，
* 用于防止发送方过多地发送数据，从而导致接收方过载。
* 当通道变得不可写时，这可能意味着接收方的缓冲区已满，
* 因此发送方需要停止发送数据，等待通道变得可写时再继续发送
* @throws Exception
*/
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception{
// 通道变得可写，可以继续发送数据
log.info("Channel is writable, resuming data transmission.");
}
/**
* 广播数据十六进制,每两个字符为一组，
*如：09 57 79 30 30 30 30 30 30 30 30 30 30 30 41 30 30 35 30 30
* 再转为对应ASCII码,即可得到明文的uuid
* @param uuid
*/
public String hexToASCII(String uuid) {
StringBuilder sb = new StringBuilder();
for (int i = 0; i < uuid.length(); i += 2) {
String hex = uuid.substring(i, i + 2);
int decimal = Integer.parseInt(hex, 16);
char asciiChar = (char) decimal;
sb.append(asciiChar);
}
String result = sb.toString();
return result;
}
/**
* 将时间戳转换为yyyy-MM-dd HH:mm:ss格式的日期字符串。
* @param timestamp 时间戳（以毫秒为单位）
* @return 格式化后的日期字符串
*/
public String timestampToDate(long timestamp) {
Instant instant = Instant.ofEpochSecond(timestamp); // 将时间戳转换为Instant对象
LocalDateTime dateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault()); // 将Instant对象转换为本地日期时间对象
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); // 定义日期格式
// 将日期时间对象格式化为字符串
String time = dateTime.format(formatter);
return time;
}
/**
* 将LocalDateTime转换为yyyy-MM-dd HH:mm:ss格式的日期字符串。
* @return 格式化后的日期字符串
*/
public String LocalDateTimeToStr() {
// 获取当前日期时间
LocalDateTime now = LocalDateTime.now();
// 定义日期时间的格式
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
// 将LocalDateTime格式化为字符串
String formattedDateTime = now.format(formatter);
return formattedDateTime;
}
}
package com.ruoyi.iot.server.tcp;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
/**
* 通道初始化
*/
public class BootNettyChannelInitializer<SocketChannel> extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) throws Exception {
//添加编解码，此处代码为解析tcp传过来的参数，为UTF-8格式，可以自定义解码格式
// ChannelOutboundHandler，依照逆序执行
ch.pipeline().addLast("encoder", new StringEncoder());
// 属于ChannelInboundHandler，依照顺序执行
ch.pipeline().addLast("decoder", new StringDecoder());
/**
* 自定义ChannelInboundHandlerAdapter
*/
ch.pipeline().addLast(new BootNettyChannelInboundHandlerAdapter());
}
}
package com.ruoyi.iot.server.tcp;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.bootstrap.ServerBootstrapConfig;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class BootNettyServer {
private static EventLoopGroup bossGroup ;
private static EventLoopGroup workerGroup;
private static ServerBootstrap serverBootstrap;
@Async
public void bind(int port) throws Exception {
/**
* 配置服务端的NIO线程组
* NioEventLoopGroup 是用来处理I/O操作的Reactor线程组
* bossGroup：用来接收进来的连接，workerGroup：用来处理已经被接收的连接,进行socketChannel的网络读写，
* bossGroup接收到连接后就会把连接信息注册到workerGroup
* workerGroup的EventLoopGroup默认的线程数是CPU核数的二倍
*/
bossGroup = new NioEventLoopGroup(1);
workerGroup = new NioEventLoopGroup();
try {
/**
* ServerBootstrap 是一个启动NIO服务的辅助启动类
*/
serverBootstrap = new ServerBootstrap();
/**
* 设置group，将bossGroup， workerGroup线程组传递到ServerBootstrap
*/
serverBootstrap = serverBootstrap.group(bossGroup, workerGroup);
/**
* ServerSocketChannel是以NIO的selector为基础进行实现的，用来接收新的连接，这里告诉Channel通过NioServerSocketChannel获取新的连接
*/
serverBootstrap = serverBootstrap.channel(NioServerSocketChannel.class);
/**
* option是设置 bossGroup，childOption是设置workerGroup
* netty 默认数据包传输大小为1024字节, 设置它可以自动调整下一次缓冲区建立时分配的空间大小，避免内存的浪费最小初始化最大 (根据生产环境实际情况来定)
* 使用对象池，重用缓冲区
*/
serverBootstrap = serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576));
serverBootstrap = serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576));
/**
* 设置 I/O处理类,主要用于网络I/O事件，记录日志，编码、解码消息
*/
serverBootstrap = serverBootstrap.childHandler(new BootNettyChannelInitializer<SocketChannel>());
log.info("netty server start success!" + "端口号："+port);
/**
* 绑定端口，同步等待成功
*/
ChannelFuture f = serverBootstrap.bind(port).sync();
/**
* 等待服务器监听端口关闭
*/
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
} finally {
/**
* 退出，释放线程池资源
*/
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
//关闭服务器
public void serverClose(){
/**
* 退出，释放线程池资源
*/
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
log.info("关闭服务");
}
//获取服务状态
public ServerBootstrapConfig serverStatus(){
return serverBootstrap.config();
}
}
package com.ruoyi.iot.server.tcp;
import cn.hutool.core.util.ObjectUtil;
import io.netty.channel.Channel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
public class UserChannelMap {
//建立用户id与netty关联
private static Map<String, Channel> userChannelMap;
static {
userChannelMap = new HashMap<String, Channel>();
}
//建立连接
public static void put(String userid, Channel channel) {
if(ObjectUtil.isNotEmpty(userChannelMap.get(userid))){
System.out.println("[设备已经上线]");
return;
}
userChannelMap.put(userid, channel);
System.out.println("[建立关联]");
print();
}
//移除
public static void remove(String userid) {
userChannelMap.remove(userid);
}
//打印通道信息
public static void print(){
for(String u : userChannelMap.keySet()){
System.out.println("[用户id]: "+u+" [通道是]: "+userChannelMap.get(u).id());
}
}
//根据通道关闭
public static void removechannel(String channelId){
if(ObjectUtil.isEmpty(channelId)){
return;
}
for (String s : userChannelMap.keySet()) {
Channel channel = userChannelMap.get(s);
if(channelId.equals(channel.id().asLongText())){
System.out.println("设备断开"+s);
userChannelMap.remove(s);
break;
}
}
}
public static Channel get(String friendid) {
print();
return userChannelMap.get(friendid);
}
/**
* 获得在线用户
* */
public static Integer getOnLine(){
return userChannelMap.size();
}
public static boolean findChannel(Channel channel){
Iterator<Map.Entry<String, Channel>> it = userChannelMap.entrySet().iterator();
boolean bool = false;
while(it.hasNext()){
Map.Entry<String, Channel> next = it.next();
Channel channelExit = next.getValue();
if(channel.equals(channelExit)){
bool = true;
break;
}
}
return bool;
}
}