接收mqtt数据高并发处理

zhuguifei

2 天以前 4a60ced80b215fcb2e2d4664b20cd744313ccc10

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttSampleCallback.java

@@ -6,6 +6,7 @@
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.poi.ss.formula.functions.T;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
@@ -42,7 +43,6 @@
public class MqttSampleCallback implements MqttCallback {
    @Value(value = "${jeecg.mqtt.role}")
    private String role;



    @Autowired
@@ -88,7 +88,7 @@
                String message = new String(mqttMessage.getPayload());
                JSONObject messageJson = JSONObject.parseObject(message);

                if (topic.startsWith("$SYS/brokers/") && topic.endsWith("connected")) {
                if (topic.startsWith("$SYS/brokers/") && topic.endsWith("connected") && !topic.endsWith("disconnected")) {
                    JSONObject client = (JSONObject) redisUtil.get(String.format(MqttConstant.MQTT_ONLINE_CLIENT, messageJson.get("clientid").toString().split("-")[1]) + messageJson.get("clientid"));
                    if (client == null) {
                        JSONObject item = new JSONObject();
@@ -102,7 +102,7 @@
                        String clientid = messageJson.getString("clientid");
                        item.put("clientid", clientid);
                        // 不符合的设备不进行管理
                        if(!clientid.matches("^[^-]+-[^-]+-[^-]+$"))  return;
                        if (!clientid.matches("^[^-]+-[^-]+-[^-]+$")) return;
                        //是否连接
                        item.put("connected", true);
                        //根据clientid解析(注意配置文件中clientid格式  例：client-1000)
@@ -113,6 +113,12 @@
                            item.put("code", info[2]);
                            redisUtil.hset(String.format(MqttConstant.MQTT_ONLINE_CLIENT, item.get("tenantId")), clientid, item);
                            System.err.println(String.format("设备: %s上线", clientid));

                            // 推送到移动端
                            String recTopic = String.format(MqttConstant.SERVICE_BROADCAST_TENANT_UPDATE_EQU_STATU, item.get("tenantId"));
                            MqMessage<JSONObject> mqMessage = new MqMessage<>(item, item.get("tenantId").toString(), recTopic);
                            sendMqttMessage(MqttConstant.SERVICE_BROADCAST_TENANT_UPDATE_EQU_STATU, mqMessage, 1);

                        } catch (Exception e) {
                            e.printStackTrace();
                        }
@@ -124,8 +130,23 @@
                if (topic.startsWith("$SYS/brokers/") && topic.endsWith("disconnected")) {
                    try {
                        String clientid = messageJson.getString("clientid");
                        redisUtil.hdel(String.format(MqttConstant.MQTT_ONLINE_CLIENT, clientid.split("-")[1]), clientid);
                        // 不符合的设备不进行管理
                        if (!clientid.matches("^[^-]+-[^-]+-[^-]+$")) return;
                        String tenantId = clientid.split("-")[1];
                        redisUtil.hdel(String.format(MqttConstant.MQTT_ONLINE_CLIENT, tenantId), clientid);
                        System.err.println(String.format("设备: %s下线", clientid));

                        //推送到移动端
                        JSONObject item = new JSONObject();
                        String[] info = clientid.split("-");
                        item.put("type", info[0]);
                        item.put("tenantId", info[1]);
                        item.put("code", info[2]);
                        item.put("connected", false);
                        String recTopic = String.format(MqttConstant.SERVICE_BROADCAST_TENANT_UPDATE_EQU_STATU, tenantId);
                        MqMessage<JSONObject> mqMessage = new MqMessage<>(item, tenantId, recTopic);
                        sendMqttMessage(MqttConstant.SERVICE_BROADCAST_TENANT_UPDATE_EQU_STATU, mqMessage, 1);

                    } catch (Exception e) {
                        e.printStackTrace();
                    }
@@ -177,38 +198,55 @@
        switch (topic) {
            // 查询设备在线
            case MqttConstant.MOBILE_QUERY_EQU_STATU:
                System.err.println("admin收到" + topic);
                // 根据设备id查询设备mqtt在线状态
                String clientId = messageJson.getString("clientId");
                JSONObject client = (JSONObject) redisUtil.hget(String.format(MqttConstant.MQTT_ONLINE_CLIENT, clientId.split("-")[1]), clientId);
                log.info("admin收到MQTT请求，topic: {}", topic);  // 改用更规范的日志记录

                ThreadUtil.execute(() -> {
                try {

                    if (client == null || client.isEmpty()) {
                        JSONObject res = new JSONObject();
                        res.put("success", false);
                        res.put("msg", "查询失败");
                        try {
                            MqttMessage sendMessage = new MqttMessage(res.toJSONString().getBytes());
                            sendMessage.setQos(0);
                            mqttUtil.getMqttClient().publish(String.format(MqttConstant.SERVICE_RES_EQU_STATU, req), sendMessage);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    // 1. 参数提取
                    String clientId = messageJson.getString("clientId");
                    if (StringUtils.isEmpty(clientId)) {
                        return;
                    }
                    String deviceKey = clientId.split("-")[1];  // 提取设备标识

                    client.put("success", true);
                    client.put("msg", "查询成功");
                    try {
                        MqttMessage sendMessage = new MqttMessage(client.toJSONString().getBytes());
                        sendMessage.setQos(0);
                        mqttUtil.getMqttClient().publish(String.format(MqttConstant.SERVICE_RES_EQU_STATU, req), sendMessage);
                        baseCommonService.addLog(client.toString(), CommonConstant.LOG_TYPE_MQTT, CommonConstant.OPERATE_MQTT_2);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
                    // 2. 查询设备状态
                    String redisKey = String.format(MqttConstant.MQTT_ONLINE_CLIENT, deviceKey);
                    JSONObject client = (JSONObject) redisUtil.hget(redisKey, clientId);

                    // 3. 异步处理响应
                    ThreadUtil.execute(() -> {
                        JSONObject response = new JSONObject();

                        // 3.1 处理查询结果
                        if (client == null || client.isEmpty()) {
                            response.put("success", false);
                            response.put("msg", "查询失败，设备不存在或离线");
                        } else {
                            response = client;  // 复用查询结果
                            response.put("success", true);
                            response.put("msg", "查询成功");
                        }

                        // 3.2 发送MQTT响应
                        try {
                            String resTopic = String.format(MqttConstant.SERVICE_RES_EQU_STATU, req);
                            MqMessage<JSONObject> mqMessage = new MqMessage<>(
                                    response,
                                    response.getString("tenantId"),
                                    resTopic
                            );

                            sendMqttMessage(resTopic, mqMessage, 2);
                            log.debug("设备状态响应发送成功: {}", response);

                        } catch (Exception e) {
                            log.error("MQTT响应发送失败", e);
                        }
                    });

                } catch (Exception e) {
                    log.error("处理设备状态查询异常", e);
                }
                break;

            // 接收设备实时数据 TODO 20250718暂不使用，使用TENANT_UP_PREFIX_REALTIME_DATA_EQP
@@ -227,10 +265,16 @@
            case MqttConstant.TENANT_UP_PREFIX_REALTIME_DATA_EQP:
                ThreadUtil.execute(() -> {
                    try {

                        RealTimeDataParentVo vo = JSON.parseObject(message, RealTimeDataParentVo.class);
                        synchronized (realTimeDataService) {
                            realTimeDataService.realTimeDataHandle(vo);
                        }
                        // 向各租户移动端发送数据
                        String recTopic = String.format(MqttConstant.SERVICE_BROADCAST_TENANT_REAL_DATA, vo.getTenantid());
                        MqMessage<RealTimeDataVo> mqMessage = new MqMessage<>(vo.getRealTime(), vo.getTenantid() + "", recTopic);
                        sendMqttMessage(MqttConstant.SERVICE_BROADCAST_TENANT_REAL_DATA, mqMessage, 1);


                         realTimeDataService.realTimeDataHandle(vo);

                    } catch (Exception e) {
                        e.printStackTrace();
                    }