将码云版本提交至gitlab -- 20241129

zhuguifei

2024-11-29 339515558253d776769dc2e2560bbb4a0450c989

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

old mode 100644
new mode 100755
@@ -3,6 +3,7 @@
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
@@ -14,12 +15,14 @@
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.*;

/**
 * mqtt
 */
@Data
@Slf4j
@Configuration
public class MqttConfig {
@@ -42,6 +45,8 @@
  private RedisUtil redisUtil;
  @Autowired
  private EmqxApi emqxApi;
  @Autowired
  private RedisTemplate redisTemplate;


  @Bean
@@ -92,6 +97,16 @@
            //订阅移动端上行指令
            mqttClient.subscribe(MqttConstant.MOBILE_UP);
            System.err.println("admin订阅" + MqttConstant.MOBILE_UP);
            // 订阅租户实时数据
            mqttClient.subscribe(MqttConstant.TENANT_UP_PREFIX_REALTIME_DATA);
            System.out.println("admin订阅" + MqttConstant.TENANT_UP_PREFIX_REALTIME_DATA);
            // 订阅租户报警数据
            mqttClient.subscribe(MqttConstant.TENANT_UP_PREFIX_REAL_FAULT_DATA);
            System.out.println("admin订阅" + MqttConstant.TENANT_UP_PREFIX_REAL_FAULT_DATA);
            mqttClient.subscribe(MqttConstant.TENANT_UP_PREFIX_FAULT_DATA);
            System.out.println("admin订阅" + MqttConstant.TENANT_UP_PREFIX_FAULT_DATA);
            mqttClient.subscribe(MqttConstant.TENANT_UP_PREFIX_EQU);
            System.out.println("admin订阅" + MqttConstant.TENANT_UP_PREFIX_EQU);
            initClients();

            break;
@@ -99,6 +114,7 @@
          case "user":
            //普通客户端只需订阅自身相关消息
            mqttClient.subscribe(MqttConstant.SERVICE_DOWN_PREFIX + "/" + mqttClientId + "/#");
            mqttClient.subscribe(MqttConstant.SERVICE_REQ_PREFIX);
            System.err.println("user订阅" + MqttConstant.SERVICE_DOWN_PREFIX + "/" + mqttClientId + "/#");
            break;

@@ -122,7 +138,7 @@
      @Override
      public void run() {
        // 在这里编写定时执行的任务逻辑
        System.out.println("定时任务执行：" + new java.util.Date());
        //System.out.println("定时任务执行：" + new java.util.Date());
        if (mqttUtil.getMqttClient() == null || !mqttUtil.getMqttClient().isConnected()) {
          try {
            conn();
@@ -142,7 +158,14 @@
   * 服务端（admin角色）启动时查询所有设备并缓存到redis
   */
  private void initClients() {
    redisUtil.del(MqttConstant.MQTT_ONLINE_CLIENT);
    //初始化时先删除所有在线设备
    Set keys = redisTemplate.keys( String.format(MqttConstant.MQTT_ONLINE_CLIENT,"*"));
    if (keys != null && !keys.isEmpty()) {
      keys.forEach(key -> System.out.println("初始化删除在线设备: " + key));
      redisTemplate.delete(keys);
    } else {
      System.out.println("初始化无在线设备: " + MqttConstant.MQTT_ONLINE_CLIENT);
    }

    JSONObject clients = emqxApi.queryEmqx(EmqxApi.CMD_CLIENTS);
    //TODO 根据emqx返回编写实体类
@@ -163,15 +186,21 @@
        //是否连接
        Boolean connected = obj.getBoolean("connected");
        item.put("connected", connected);
        //
        //根据clientid解析(注意配置文件中clientid格式  例：client-1000)
        try {
        String[] info = clientid.split("-");
        item.put("type", info[0]);
        item.put("tenantId", info[1]);
        item.put("code", info[2]);
          //item.put("code", info[2]);

        if (connected) {
          redisUtil.set(MqttConstant.MQTT_ONLINE_CLIENT + clientid, item);
            redisUtil.hset(String.format(MqttConstant.MQTT_ONLINE_CLIENT,item.get("tenantId").toString()) , clientid, item);
        }
        }catch (Exception e){
          e.printStackTrace();
        }




      }