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.AmqpMessageListener;
|
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.dto.CommonReportMessage;
|
import com.sandu.ximon.admin.manager.iot.rrpc.enums.A5OrderEnum;
|
import com.sandu.ximon.admin.manager.iot.rrpc.util.FrameUtils;
|
import com.sandu.ximon.admin.utils.RedisUtils;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.stereotype.Component;
|
|
import javax.jms.Message;
|
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_msg")
|
public class MsgMqttCallBack extends AbsMqttCallBack {
|
|
private static final String localMqttConnectTypeOfSync = "1";
|
|
private static final String localMqttConnectTypeOfAsync = "2";
|
|
public static final String localMqttSyncFrame = "local_mqtt_sync_frame:";
|
|
|
protected final static ExecutorService EXECUTOR_SERVICE = new ThreadPoolExecutor(
|
Runtime.getRuntime().availableProcessors(),
|
Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
|
new LinkedBlockingQueue<>(50000), new MsgMqttCallBack.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("接收到消息---MsgMqttCallBack:topic={},message={}", topic, message);
|
}
|
|
|
/**
|
* 在这里处理您收到消息后的具体业务逻辑。
|
*/
|
private void processMessage(String topic,String messageString) {
|
try {
|
|
String mac = topic.split("/")[3];
|
LocalMqttMsg localMqttMsg = JSON.parseObject(messageString, LocalMqttMsg.class);
|
|
if (localMqttMsg.getConnectType().equals(localMqttConnectTypeOfAsync)){
|
// 设备数据上报
|
processTask(null,mac, localMqttMsg);
|
}
|
else if (localMqttMsg.getConnectType().equals(localMqttConnectTypeOfSync)){
|
System.out.println(localMqttMsg.getPayload());
|
MQTT_RETURN_FRAME_MAP.put(mac,localMqttMsg.getPayload());
|
}
|
|
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
|
/**
|
* 处理任务
|
*
|
* @param productKey 产品码
|
* @param deviceName 产品名称
|
* @param localMqttMsg 上报消息localMqttMsg
|
*/
|
private void processTask(String productKey, String deviceName, LocalMqttMsg localMqttMsg) {
|
if (localMqttMsg == null) {
|
return;
|
}
|
log.info("处理订阅");
|
log.info(localMqttMsg.toString());
|
CommonFrame frame = HexFrameUtils.transformMessageToFrame(localMqttMsg.getPayload());
|
if (frame.getOrderType().equals(A5OrderEnum.RESPONSE_LIGHT_DATA.getCode())) {
|
// 单灯数据上报处理
|
LightDataProcessor.getInstance().process(productKey, deviceName, frame);
|
} else if (frame.getOrderType().equals(A5OrderEnum.RESPONSE_C3_DATA.getCode())) {
|
// C3充电桩上报处理
|
c3ChargingProcessor.c3ChargingProcessorgetInstance().process(productKey, deviceName, frame);
|
} else if (frame.getOrderType().equals(A5OrderEnum.RESPONSE_ATMOSPHERE_DATA.getCode())) {
|
// 大气数据指令上报
|
AirDataProcessor.getInstance().process(productKey, deviceName, frame);
|
} else if (frame.getOrderType().equals(A5OrderEnum.RESPONSE_POLE_MONITOR_DATA.getCode())) {
|
PoleMonitorDataProcessor.getInstance().process(productKey, deviceName, frame);
|
}
|
}
|
}
|
