package org.jeecg.modules.iot.mqtt.callback;
|
|
import com.alibaba.fastjson.JSON;
|
import com.alibaba.fastjson.JSONObject;
|
import lombok.extern.slf4j.Slf4j;
|
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
|
import org.eclipse.paho.client.mqttv3.MqttCallback;
|
import org.eclipse.paho.client.mqttv3.MqttClient;
|
import org.eclipse.paho.client.mqttv3.MqttMessage;
|
import org.jeecg.common.util.DateUtils;
|
import org.jeecg.modules.iot.entity.MqttParameter;
|
import org.jeecg.modules.iot.entity.ServerDeploy;
|
import org.jeecg.modules.iot.mqtt.config.MqttConfiguration;
|
import org.jeecg.modules.iot.mqtt.config.MqttCustomerClient;
|
import org.jeecg.modules.iot.service.IServerDeployService;
|
import org.jeecg.modules.message.websocket.WebSocket;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.context.annotation.Lazy;
|
import org.springframework.stereotype.Component;
|
|
import java.nio.charset.StandardCharsets;
|
import java.util.List;
|
|
|
/**
|
* 消费监听
|
*
|
* @author: cuikaidong
|
* @create: 2023-09-27 11:36
|
*/
|
@Slf4j
|
@Component
|
public class PushCallback implements MqttCallback {
|
|
private static MqttClient client;
|
@Lazy
|
@Autowired
|
private MqttConfiguration mqttConfiguration;
|
@Autowired
|
private IServerDeployService serverDeployService;
|
@Autowired
|
private WebSocket webSocket;
|
|
@Override
|
public void connectionLost(Throwable throwable) {
|
log.error("MQTT连接已断开", throwable);
|
|
int retryCount = 0;
|
final long baseDelay = 1000; // 初始延迟1秒
|
final long maxDelay = 60000; // 最大延迟60秒（避免延迟过长）
|
final double backoffFactor = 2.0; // 指数退避因子
|
|
while (true) { // 无限循环，直到连接成功
|
try {
|
// 计算退避延迟，使用Math.min确保不超过最大延迟
|
long delay = (long) Math.min(baseDelay * Math.pow(backoffFactor, retryCount), maxDelay);
|
if (retryCount > 0) {
|
log.info("第{}次重连尝试将在{}ms后进行", retryCount + 1, delay);
|
Thread.sleep(delay);
|
}
|
|
log.info("开始第{}次重连尝试", retryCount + 1);
|
|
MqttCustomerClient mqttCustomerClient = mqttConfiguration.getMqttCustomerClient();
|
|
// 尝试连接
|
mqttCustomerClient.connect(
|
mqttConfiguration.getHost(),
|
mqttConfiguration.getClientId(),
|
mqttConfiguration.getUsername(),
|
mqttConfiguration.getPassword(),
|
mqttConfiguration.getTimeout(),
|
mqttConfiguration.getKeepalive()
|
);
|
|
// 订阅主题
|
log.info("MQTT连接已恢复，正在重新订阅主题");
|
|
List<ServerDeploy> serverDeploys = serverDeployService.list();
|
String[] deviceTopics = serverDeploys.stream()
|
.map(deploy -> "IOT\\" + deploy.getServerCode() + "\\Mutually2")
|
.toArray(String[]::new);
|
|
if (deviceTopics.length > 0) {
|
mqttCustomerClient.subscribe(deviceTopics);
|
log.info("已重新订阅{}个设备主题", deviceTopics.length);
|
}
|
|
String[] systemTopics = {"$SYS/brokers/+/clients/#"};
|
mqttCustomerClient.subscribe(systemTopics);
|
log.info("已重新订阅系统主题");
|
|
log.info("MQTT连接已完全恢复，共尝试{}次", retryCount + 1);
|
return; // 连接和订阅成功，退出循环
|
} catch (Exception e) {
|
retryCount++;
|
log.error("第{}次重连尝试失败: {}", retryCount, e.getMessage(), e);
|
}
|
}
|
}
|
/**
|
* mqtt订阅数据
|
*
|
* @param topic
|
* @param message
|
* @throws Exception
|
*/
|
@Override
|
public void messageArrived(String topic, MqttMessage message) throws Exception {
|
// 订阅数据
|
// 通过topic区分消息
|
// 关键修改：显式指定UTF-8编码
|
String content = new String(message.getPayload(), StandardCharsets.UTF_8);
|
JSONObject jsonObject = JSON.parseObject(content);
|
String clientId = String.valueOf(jsonObject.get("id"));
|
if (clientId.equals("null")) {
|
clientId = String.valueOf(jsonObject.get("clientid"));
|
}
|
ServerDeploy serverDeploy = serverDeployService.findByServerCode(clientId);
|
if (serverDeploy == null) return;
|
// 客户端已掉线
|
if (topic.endsWith("disconnected")) {
|
// 修改守护和采集连接状态
|
updateCollectState(topic, serverDeploy);
|
// 客户端已上线
|
} else if (topic.endsWith("connected")) {
|
// 修改守护和采集连接状态
|
updateCollectState(topic, serverDeploy);
|
} else if (topic.endsWith("Mutually2")) {
|
receiveCollectMessage(jsonObject, serverDeploy);
|
}
|
}
|
|
@Override
|
public void deliveryComplete(IMqttDeliveryToken token) {
|
// log.info("deliveryComplete---------" + token.isComplete());
|
}
|
|
/**
|
* 修改守护和采集连接状态
|
*
|
* @param topic
|
* @param serverDeploy
|
*/
|
void updateCollectState(String topic, ServerDeploy serverDeploy) {
|
JSONObject obj = new JSONObject();
|
obj.put("cmd", "collect");//业务类型
|
obj.put("id", serverDeploy.getServerCode());//消息id
|
if (topic.endsWith("disconnected")) {
|
serverDeploy.setGuardState(0);
|
serverDeploy.setCollectState(0);
|
obj.put("message1", 0);
|
obj.put("message", 0);
|
} else {
|
serverDeploy.setGuardState(1);
|
obj.put("message1", 1);
|
}
|
webSocket.sendAllMessage(obj.toJSONString());
|
serverDeployService.updateById(serverDeploy);
|
}
|
|
/**
|
* 采集消息解析
|
*
|
* @param jsonObject
|
*/
|
void receiveCollectMessage(JSONObject jsonObject, ServerDeploy serverDeploy) {
|
String clientId = String.valueOf(jsonObject.get("id"));
|
String type = String.valueOf(jsonObject.get("type"));
|
String time = String.valueOf(jsonObject.get("time"));
|
MqttParameter mqttParameter = new MqttParameter();
|
String parameter1 = String.valueOf(jsonObject.get("parameter1"));
|
String parameter2 = String.valueOf(jsonObject.get("parameter2"));
|
String parameter3 = String.valueOf(jsonObject.get("parameter3"));
|
String parameter4 = String.valueOf(jsonObject.get("parameter4"));
|
String parameter5 = String.valueOf(jsonObject.get("parameter5"));
|
JSONObject obj = new JSONObject();
|
obj.put("cmd", type);//业务类型
|
obj.put("id", clientId);//消息id
|
switch (type) {
|
// 心跳接收
|
case "palpitate":
|
// log.info("心跳状态接收! 服务器名称:" + serverDeploy.getServerName() + " 时间:" + DateUtils.getTimestamp());
|
break;
|
case "version":
|
// 展示采集软件配置文件版本 ,修改采集软件版本号
|
if (parameter1 != null && !parameter1.equals("null") && !parameter1.equals("0")) {
|
obj.put("state", parameter1);
|
// 采集软件
|
serverDeploy.setNewCollectVersion(parameter1);
|
}
|
if (parameter2 != null && !parameter2.equals("null") && !parameter2.equals("0")) {
|
obj.put("message", parameter2);
|
// 配置文件
|
serverDeploy.setNewDeployVersion(parameter2);
|
}
|
serverDeployService.updateById(serverDeploy);
|
//采集软件消息内容
|
//配置文件消息内容
|
log.info("守护服务连接成功! 服务器名称:" + serverDeploy.getServerName() + " 时间:" + DateUtils.getTimestamp());
|
break;
|
case "log":
|
// 日志
|
obj.put("state", parameter4);//消息内容
|
obj.put("message", parameter5);//消息内容
|
log.info("日志上传接收! 服务器名称:" + serverDeploy.getServerName() + " 时间:" + DateUtils.getTimestamp());
|
break;
|
case "start":
|
// 启动采集软件
|
obj.put("state", parameter2);
|
obj.put("message", parameter3);
|
int collectState = Integer.parseInt(parameter1);
|
serverDeploy.setCollectState(collectState);
|
if (parameter2.equals("1")) {
|
serverDeployService.updateById(serverDeploy);
|
}
|
log.info("启动采集软件接收! 服务器名称:" + serverDeploy.getServerName() + " 时间:" + DateUtils.getTimestamp());
|
break;
|
case "collect":
|
// 采集状态
|
obj.put("message", parameter1);
|
if (!serverDeploy.getCollectState().toString().equals(parameter1)) {
|
serverDeploy.setCollectState(Integer.getInteger(parameter1));
|
serverDeployService.updateById(serverDeploy);
|
}
|
log.info("心跳♥采集状态接收! 服务器名称:" + serverDeploy.getServerName() + " 时间:" + DateUtils.getTimestamp());
|
break;
|
default:
|
return;
|
}
|
if (obj.size() > 2) {
|
webSocket.sendAllMessage(obj.toJSONString());
|
}
|
}
|
}
|
