package com.sandu.ximon.admin.localMQTT.callback;
|
|
|
import com.alibaba.fastjson.JSON;
|
import com.sandu.ximon.admin.localMQTT.model.LocalMqttMsg;
|
import com.sandu.ximon.admin.localMQTT.util.HexFrameUtils;
|
import com.sandu.ximon.admin.manager.iot.amqp.processor.AirDataProcessor;
|
import com.sandu.ximon.admin.manager.iot.amqp.processor.LightDataProcessor;
|
import com.sandu.ximon.admin.manager.iot.amqp.processor.PoleMonitorDataProcessor;
|
import com.sandu.ximon.admin.manager.iot.amqp.processor.c3ChargingProcessor;
|
import com.sandu.ximon.admin.manager.iot.rrpc.dto.CommonFrame;
|
import com.sandu.ximon.admin.manager.iot.rrpc.enums.A5OrderEnum;
|
import com.sandu.ximon.admin.utils.RedisUtils;
|
import lombok.extern.slf4j.Slf4j;
|
import org.eclipse.paho.client.mqttv3.MqttClient;
|
import org.springframework.stereotype.Component;
|
|
import java.util.Map;
|
import java.util.concurrent.*;
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
import static com.sandu.ximon.admin.localMQTT.util.MqttClientUtil.MQTT_RETURN_FRAME_MAP;
|
|
/**
|
* @author van
|
* @version 1.0
|
* msg：默认回调
|
* @date 2022/11/9 16:24
|
*/
|
@Slf4j
|
@Component("java_server_status")
|
public class StatusMqttCallBack extends AbsMqttCallBack {
|
|
private static final String localMqttConnectStatusConnected = "connected";
|
|
private static final String localMqttConnectStatusDisconnected = "disconnected";
|
|
public static final String localMqttConnectStatus = "localMqttConnectStatus.";
|
|
|
|
protected final static ExecutorService EXECUTOR_SERVICE = new ThreadPoolExecutor(
|
Runtime.getRuntime().availableProcessors(),
|
Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
|
new LinkedBlockingQueue<>(50000), new StatusMqttCallBack.NameTreadFactory());
|
|
static class NameTreadFactory implements ThreadFactory {
|
|
private final AtomicInteger mThreadNum = new AtomicInteger(1);
|
|
@Override
|
public Thread newThread(Runnable r) {
|
return new Thread(r, "local-MQTT-msg-thread-" + mThreadNum.getAndIncrement());
|
}
|
}
|
|
@Override
|
protected void handleReceiveMessage(String topic, String message) {
|
EXECUTOR_SERVICE.submit(() -> processMessage(topic,message));
|
|
// log.info("接收到消息---StatusMqttCallBack:topic={},message={}", topic, message);
|
}
|
|
|
/**
|
* 在这里处理您收到消息后的具体业务逻辑。
|
*/
|
private void processMessage(String topic,String messageString) {
|
try {
|
|
String mac = topic.split("/")[4];
|
String status = topic.split("/")[5];
|
|
if (status.equals(localMqttConnectStatusConnected)){
|
// 设备数据上报
|
boolean set = RedisUtils.getBean().set(localMqttConnectStatus + mac, 1);
|
log.error("设备上线------，mac:{},上线,{}",mac,set);
|
}
|
else if (status.equals(localMqttConnectStatusDisconnected)){
|
boolean set = RedisUtils.getBean().set(localMqttConnectStatus+mac,0);
|
log.error("设备下线------，mac:{},下线.{}",mac,set);
|
}
|
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
|
}
|
