add:接入mqtt2

mes/hd/nladmin-system/pom.xml

@@ -251,16 +251,17 @@
             <artifactId>mybatis-plus-boot-starter</artifactId>
             <version>3.4.0</version>
         </dependency>
-        <dependency>
-            <groupId>org.apache.velocity</groupId>
-            <artifactId>velocity-engine-core</artifactId>
-            <version>2.3</version>
-        </dependency>
         <dependency>
             <groupId>com.baomidou</groupId>
             <artifactId>mybatis-plus-generator</artifactId>
             <version>3.4.0</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.velocity</groupId>
+            <artifactId>velocity-engine-core</artifactId>
+            <version>2.3</version>
+        </dependency>
+
         <dependency>
             <groupId>xerces</groupId>
             <artifactId>xercesImpl</artifactId>

wcs/nladmin-system/pom.xml

@@ -237,6 +237,12 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
+        <!-- https://mvnrepository.com/artifact/com.alibaba/transmittable-thread-local -->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>transmittable-thread-local</artifactId>
+            <version>2.14.2</version>
+        </dependency>
 
         <!--Spring boot 测试-->
         <dependency>
@@ -409,19 +415,6 @@
             <version>1.6.2</version>
         </dependency>
 
-        <!-- 解析客户端操作系统、浏览器信息 -->
-<!--        <dependency>-->
-<!--            <groupId>nl.basjes.parse.useragent</groupId>-->
-<!--            <artifactId>yauaa</artifactId>-->
-<!--            <version>5.23</version>-->
-<!--        </dependency>-->
-
-<!--        <dependency>-->
-<!--            <groupId>eu.bitwalker</groupId>-->
-<!--            <artifactId>UserAgentUtils</artifactId>-->
-<!--            <version>1.21</version>-->
-<!--        </dependency>-->
-
         <!-- Lucence核心包 -->
         <dependency>
             <groupId>org.apache.lucene</groupId>
@@ -468,9 +461,20 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator</artifactId>
         </dependency>
-
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-stream</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-mqtt</artifactId>
+        </dependency>
     </dependencies>
-
+    <!--mqtt依赖-->
     <distributionManagement>
         <!--正式版本-->
         <repository>

wcs/nladmin-system/src/main/java/org/nl/acs/log/LokiLogAspect.java

@@ -38,10 +38,6 @@ public class LokiLogAspect {
      */
     @Around("operatorLog()")
     public  Object around(ProceedingJoinPoint pjp) throws Throwable {
-//        ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
-//        HttpServletRequest request = attributes.getRequest();
-//        HttpServletResponse response = attributes.getResponse();
-
         Signature signature = pjp.getSignature();
         MethodSignature methodSignature = (MethodSignature) signature;
         Method method = methodSignature.getMethod();

wcs/nladmin-system/src/main/java/org/nl/common/utils/YmlConfigFileUtil.java

@@ -0,0 +1,28 @@
+package org.nl.common.utils;
+
+import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+
+import java.util.Properties;
+
+/**
+ * @Author: lyd
+ * @Description: 配置文件获取方法
+ * @Date: 2023/12/6
+ */
+public class YmlConfigFileUtil {
+    public static Properties readConfig(String configFile) {
+        // 创建 Resource 对象
+        Resource resource = new ClassPathResource(configFile);
+
+        // 创建 YamlPropertiesFactoryBean
+        YamlPropertiesFactoryBean yamlPropertiesFactoryBean = new YamlPropertiesFactoryBean();
+        yamlPropertiesFactoryBean.setResources(resource);
+
+        // 获取 Properties 对象
+        Properties properties = yamlPropertiesFactoryBean.getObject();
+
+        return properties;
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/agvconfig/ProtocolCodec.java

@@ -0,0 +1,22 @@
+package org.nl.config.agvconfig;
+
+import io.netty.buffer.ByteBuf;
+
+/*
+ * @author ZZQ
+ * @Date 2024/1/3 17:18
+ */
+public class ProtocolCodec {
+
+   public static void readEmp(ByteBuf buf, int size){
+        buf.readBytes(size);
+    }
+    public static byte[] readBody(ByteBuf buf){
+        byte[] body = new byte[buf.readableBytes()];
+        buf.readBytes(body);
+        if (body[0]<<8+body[1] == 0X87CD){
+            return body;
+        }
+        return new byte[0];
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/lucene/AsyncLuceneAppender.java

@@ -0,0 +1,32 @@
+package org.nl.config.lucene;
+/**
+ * @author ldjun
+ * @version 1.0
+ * @date 2023年08月24日 13:00
+ * @desc desc
+ */
+
+import ch.qos.logback.classic.AsyncAppender;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.MDC;
+
+import java.util.Map;
+
+public class AsyncLuceneAppender extends AsyncAppender {
+
+
+    @Override
+    protected void append(ILoggingEvent event) {
+        String traceId = LuceneAppender.traceIdTL.get();
+        if (StringUtils.isNotEmpty(traceId)){
+            MDC.put("traceId",traceId);
+            Map<String, String> mdcPropertyMap = event.getMDCPropertyMap();
+            if (mdcPropertyMap.getClass().getName().contains("SynchronizedMap")){
+                mdcPropertyMap.put("traceId",traceId);
+            }
+            MDC.clear();
+        }
+        super.append(event);
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/lucene/LogMessageConstant.java

@@ -0,0 +1,46 @@
+package org.nl.config.lucene;
+
+/**
+ * @Author: lyd
+ * @Description: 定义lucene相关常量
+ * @Date: 2023/8/25
+ */
+public class LogMessageConstant {
+    /** */
+    public final static String SORT_NAME = "time";
+    /** 级别 */
+    public final static String FIELD_LEVEL = "level";
+    /** 时间 */
+    public final static String FIELD_TIMESTAMP = "timestamp";
+    /** 类的限定名 */
+    public final static String FIELD_CLASS_NAME = "logger";
+    /** 线程名 */
+    public final static String FIELD_THREAD = "thread";
+    /** 日志内容 */
+    public final static String FIELD_MESSAGE = "message";
+    public final static String FIELD_TRACEID = "tlogTraceId";
+    // 定义颜色值
+    /** 文本颜色：黑色 */
+    public final static String COLOR_BLACK = "\u001B[30m";
+    /** 文本颜色：红色 */
+    public final static String COLOR_RED = "\u001B[31m";
+    /** 文本颜色：绿色 */
+    public final static String COLOR_GREEN = "\u001B[32m";
+    /** 文本颜色：黄色 */
+    public final static String COLOR_YELLOW = "\u001B[33m";
+    /** 文本颜色：蓝色 */
+    public final static String COLOR_BLUE = "\u001B[34m";
+    /** 文本颜色：品红色 */
+    public final static String COLOR_MAGENTA = "\u001B[35m";
+    /** 文本颜色：青色 */
+    public final static String COLOR_CYAN = "\u001B[36m";
+    /** 文本颜色：白色 */
+    public final static String COLOR_WHITE = "\u001B[37m";
+    /** 文本颜色重置 */
+    public final static String COLOR_RESET = "\u001B[0m";
+    /** 背景颜色：黄色 */
+    public final static String BACKGROUND_YELLOW = "\u001B[43m";
+    /** 索引路径 */
+    public final static String INDEX_DIR = "E:\\lucene\\index";
+
+}

wcs/nladmin-system/src/main/java/org/nl/config/lucene/LuceneAppender.java

@@ -0,0 +1,99 @@
+package org.nl.config.lucene;
+/**
+ * @author ldjun
+ * @version 1.0
+ * @date 2023年08月24日 13:00
+ * @desc desc
+ */
+
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.AppenderBase;
+import com.alibaba.ttl.TransmittableThreadLocal;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.lucene.analysis.Analyzer;
+import org.apache.lucene.document.*;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.FSDirectory;
+import org.nl.common.utils.YmlConfigFileUtil;
+import org.wltea.analyzer.lucene.IKAnalyzer;
+
+import java.io.IOException;
+import java.nio.file.Paths;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+public class LuceneAppender extends AppenderBase<ILoggingEvent> {
+
+    public static final TransmittableThreadLocal<String> traceIdTL = new TransmittableThreadLocal();
+    public LuceneProperties properties;
+    public static Directory index;
+    private List<LucenePropertyAndEncoder>  encoders;
+    public static IndexWriter indexWriter;
+
+
+
+    @Override
+    public void start() {
+        super.start();
+        try {
+            // 读取配置文件
+            Properties properties = YmlConfigFileUtil.readConfig("config/application.yml");
+
+            // 获取配置值
+            String luceneDir = properties.getProperty("lucene.index.path");
+            System.out.println("---index地址----"+luceneDir);
+            index = FSDirectory.open(Paths.get(luceneDir));
+            // 初始化 Lucene 索引
+            Analyzer analyzer = new IKAnalyzer();
+            IndexWriterConfig config = new IndexWriterConfig(analyzer);
+            indexWriter = new IndexWriter(index, config);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+
+    @Override
+    protected void append(ILoggingEvent event) {
+        Document doc = new Document();
+        for (Property property : this.properties.getProperties()) {
+            LucenePropertyAndEncoder encoder = new LucenePropertyAndEncoder(property, this.context);
+            String encode = encoder.encode(event);
+            doc.add(new StringField(property.getName(), encode, Field.Store.YES));
+        }
+        Map<String, String> map = event.getMDCPropertyMap();
+        if (!map.isEmpty() && StringUtils.isNotEmpty(map.get("traceId"))){
+            doc.add(new StringField("traceId",map.get("traceId"), Field.Store.YES));
+        }else {
+            doc.add(new StringField("traceId"," ", Field.Store.YES));
+        }
+
+        doc.add(new TextField(LogMessageConstant.FIELD_MESSAGE, event.getFormattedMessage(), Field.Store.YES));
+        doc.add(new StringField(LogMessageConstant.FIELD_TIMESTAMP, String.valueOf(event.getTimeStamp()),Field.Store.YES));
+        doc.add(new NumericDocValuesField(LogMessageConstant.SORT_NAME, event.getTimeStamp()));
+        try {
+            indexWriter.addDocument(doc);
+            indexWriter.commit();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public void stop() {
+        super.stop();
+        try {
+            indexWriter.close();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void setProperties(LuceneProperties properties) {
+        this.properties = properties;
+
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/lucene/LuceneProperties.java

@@ -0,0 +1,23 @@
+package org.nl.config.lucene;
+
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class LuceneProperties {
+
+	private List<Property> properties;
+
+	public LuceneProperties() {
+		this.properties = new ArrayList<Property>();
+	}
+
+	public List<Property> getProperties() {
+		return properties;
+	}
+
+	public void addProperty(Property property) {
+		properties.add(property);
+	}
+
+}

wcs/nladmin-system/src/main/java/org/nl/config/lucene/LucenePropertyAndEncoder.java

@@ -0,0 +1,38 @@
+package org.nl.config.lucene;
+
+import ch.qos.logback.classic.PatternLayout;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.Context;
+import ch.qos.logback.core.pattern.PatternLayoutBase;
+
+/*
+ * @author ZZQ
+ * @Date 2023/12/22 18:11
+ */
+public class LucenePropertyAndEncoder {
+
+    private Property property;
+
+    private  PatternLayoutBase layout = new PatternLayout();
+
+    public LucenePropertyAndEncoder(Property property, Context context) {
+        this.property = property;
+        this.layout.setContext(context);
+        this.layout.setPattern(String.valueOf(property.getValue()));
+        this.layout.setPostCompileProcessor(null);
+        this.layout.start();
+    }
+
+    public String encode(ILoggingEvent event) {
+        return layout.doLayout(event);
+    }
+
+    public String getName() {
+        return property.getName();
+    }
+
+    public boolean allowEmpty() {
+        return property.isAllowEmpty();
+    }
+}
+

wcs/nladmin-system/src/main/java/org/nl/config/lucene/Property.java

@@ -0,0 +1,44 @@
+package org.nl.config.lucene;
+
+/*
+ * @author ZZQ
+ * @Date 2023/12/26 15:30
+ */
+public class Property {
+    private String name;
+    private String value;
+    private boolean allowEmpty;
+
+    public Property() {
+    }
+
+    public Property(String name, String value, boolean allowEmpty) {
+        this.name = name;
+        this.value = value;
+        this.allowEmpty = allowEmpty;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+    public boolean isAllowEmpty() {
+        return allowEmpty;
+    }
+
+    public void setAllowEmpty(boolean allowEmpty) {
+        this.allowEmpty = allowEmpty;
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/mqtt2/MQServer.java

@@ -0,0 +1,78 @@
+package org.nl.config.mqtt2;
+
+import org.apache.commons.lang3.StringUtils;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.nl.config.mqtt2.callback.PublishCallback;
+import org.nl.config.mqtt2.callback.SubsribeCallback;
+import org.nl.config.mqtt2.config.MqttConfig;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import java.util.UUID;
+
+/*
+ * @author ZZQ
+ * @Date 2024/1/31 14:07
+ *
+ */
+@Service
+@ConditionalOnProperty(name = "mqtt.active", havingValue = "true")
+public class MQServer {
+
+    @Autowired
+    private MqttConfig mqttConfig;
+
+    public static MqttClient subsribeClient;
+
+    public static MqttClient publishClient;
+
+    @PostConstruct
+    public void init() throws Exception {
+        this.initSubsribe();
+        this.initPublish();
+    }
+
+
+    public static void sendMsg(String topic,String body,int qos,Boolean retained){
+        try {
+            if (publishClient!=null && StringUtils.isNotEmpty(body)){
+                MqttMessage message = new MqttMessage();
+                message.setQos(qos);
+                message.setRetained(retained);
+                message.setPayload(body.getBytes());
+                publishClient.publish(topic,message);
+            }
+        }catch (Exception ex){
+            ex.printStackTrace();
+        }
+    }
+
+    public void shutdown() throws Exception {
+        subsribeClient.disconnect();
+        publishClient.disconnect();
+    }
+
+    private void initSubsribe() throws Exception {
+        if (subsribeClient!=null){
+            System.out.println("重新连接");
+            subsribeClient.disconnect();
+        }
+        subsribeClient = new MqttClient(mqttConfig.getUrl(), mqttConfig.getClientId(), new MemoryPersistence());
+        subsribeClient.connect(mqttConfig.getOption());
+        subsribeClient.setCallback(new SubsribeCallback());
+        subsribeClient.subscribe(mqttConfig.getTopics());
+    }
+    private void initPublish() throws Exception {
+        if (publishClient!=null){
+            System.out.println("重新连接");
+            publishClient.disconnect();
+        }
+        publishClient = new MqttClient(mqttConfig.getUrl(), UUID.randomUUID().toString(), new MemoryPersistence());
+        publishClient.connect(mqttConfig.getOption());
+        publishClient.setCallback(new PublishCallback());
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/mqtt2/PublishDemo.java

@@ -0,0 +1,28 @@
+package org.nl.config.mqtt2;
+
+import cn.dev33.satoken.annotation.SaIgnore;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestMapping;
+
+/*
+ * @author ZZQ
+ * @Date 2024/1/31 17:12
+ */
+//@RestController
+public class PublishDemo {
+
+    @Autowired
+    MQServer server;
+
+    @RequestMapping("/publish")
+    @SaIgnore
+    public void send(String topic,String msg){
+        MQServer.sendMsg(topic,msg,1,true);
+    }
+
+    @RequestMapping("/init")
+    @SaIgnore
+    public void d222() throws Exception {
+        server.init();
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/mqtt2/callback/PublishCallback.java

@@ -0,0 +1,29 @@
+package org.nl.config.mqtt2.callback;
+
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+/*
+ * @author ZZQ
+ * @Date 2024/1/31 13:55
+ */
+public class PublishCallback implements MqttCallback {
+
+    @Override
+    public void connectionLost(Throwable throwable) {
+        //重连，调用服务初始化init
+    }
+
+    @Override
+    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
+        //接收到消息：先放队列，根据不同topic处理不同处理逻辑
+    }
+
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+        //消息接收成功后
+        boolean complete = iMqttDeliveryToken.isComplete();
+        System.out.println("消息处理结果："+complete);
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/mqtt2/callback/SubsribeCallback.java

@@ -0,0 +1,31 @@
+package org.nl.config.mqtt2.callback;
+
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.nl.config.mqtt2.msg.MsgPoolManager;
+import org.nl.config.mqtt2.msg.MsgWorker;
+
+/*
+ * @author ZZQ
+ * @Date 2024/1/31 13:55
+ */
+public class SubsribeCallback implements MqttCallback {
+
+    @Override
+    public void connectionLost(Throwable throwable) {
+        //重连，调用服务初始化init
+    }
+
+    @Override
+    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
+        //接收到消息：先放队列，根据不同topic处理不同处理dd逻辑
+        System.out.println("接收到消息topic:"+topic+"——"+new String(mqttMessage.getPayload()));
+        MsgPoolManager.hander(new MsgWorker(topic,new String(mqttMessage.getPayload())));
+    }
+
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+        //消息接收如果处理失败允许在这里重试
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/mqtt2/config/MqttConfig.java

@@ -0,0 +1,58 @@
+package org.nl.config.mqtt2.config;
+
+import lombok.Data;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.PostConstruct;
+
+
+/*
+ * @author ZZQ
+ * @Date 2024/1/31 14:07
+ */
+@Configuration
+@ConfigurationProperties(prefix = "spring.mqtt")
+@ConditionalOnProperty(name = "mqtt.active", havingValue = "true")
+@Data
+public class MqttConfig {
+
+    private String username;
+
+    private String password;
+
+    private String url;
+
+    private String clientId;
+
+    private String[] topics;
+
+    private int timeout;
+
+    private int keepalive;
+
+    private MqttConnectOptions option;
+
+    @PostConstruct
+    public void initOption(){
+        MqttConnectOptions options = new MqttConnectOptions();
+        // 设定清除会话信息，true时，每次连接都会建立新的会话，false时，服务端会保留会话信息
+        options.setCleanSession(true);
+        //options.setCleanSession(true);
+        // 设定重连机制，设定为true时,mqtt的重连机制会启动,当mqtt client掉线之后它会进入重连
+        options.setAutomaticReconnect(true);
+        if (!username.equals("a")){
+            options.setUserName(username);
+        }
+        if (!password.equals("a")){
+            options.setPassword(password.toCharArray());
+        }
+        options.setConnectionTimeout(timeout);
+        options.setKeepAliveInterval(keepalive);
+        // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
+        //options.setWill("willTopic", WILL_DATA, 2, false);
+        option=options;
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/config/mqtt2/msg/MsgPoolManager.java

@@ -0,0 +1,78 @@
+package org.nl.config.mqtt2.msg;
+
+import org.nl.config.mqtt2.MQServer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.util.Queue;
+import java.util.concurrent.*;
+
+/*
+ * @author ZZQ
+ * @Date 2024/1/31 15:19
+ * 待确认：1.所有消走线程还是先走队列；2.同一个topic消息处理需要保证消费顺序3.消息写时订阅也会订阅毁掉会拿到自己写的信息
+ */
+@Component
+@ConditionalOnBean(value = MQServer.class)
+public class MsgPoolManager {
+
+    @Autowired
+    MQServer mqServer;
+    // 线程池维护线程的最少数量
+    private final static int CORE_POOL_SIZE = 2;
+    // 线程池维护线程的最大数量
+    private final static int MAX_POOL_SIZE = 20 ;
+    // 线程池维护线程所允许的空闲时间
+    private final static int KEEP_ALIVE_TIME = 10;
+    // 线程池 所使用的缓存队列大小
+    private final static int WORK_QUEUE_SIZE = 1000;
+
+
+    private final  static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);
+
+
+    static Queue<Runnable> msgQueue = new LinkedBlockingQueue<>();
+
+    static ThreadPoolExecutor threadPool =null;
+
+    static ScheduledFuture scheduledFuture =null;
+
+    static {
+
+    }
+
+
+    public static void hander(MsgWorker r){
+        threadPool.execute(r);
+    }
+
+    @PostConstruct
+    public void start() {
+        threadPool = new ThreadPoolExecutor(CORE_POOL_SIZE,MAX_POOL_SIZE,
+                KEEP_ALIVE_TIME, TimeUnit.SECONDS,new ArrayBlockingQueue<>(WORK_QUEUE_SIZE), (r, executor) -> msgQueue.add(r));
+
+//        scheduledFuture = scheduler.scheduleAtFixedRate(() -> {
+//            // 判断缓存队列是否存在记录
+//            System.out.println("定时任务启动"+msgQueue.size());
+//            if(!msgQueue.isEmpty()){
+//                if(threadPool.getQueue().size() < WORK_QUEUE_SIZE){
+//                    Runnable worker = msgQueue.poll();
+//                    threadPool.execute(worker);
+//                }
+//            }
+//        },0,1,TimeUnit.SECONDS);
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            System.out.println("----关闭线程池-----");
+            try {
+                mqServer.shutdown();
+                threadPool.shutdown();
+//            scheduler.shutdown();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }));
+    }
+
+}

wcs/nladmin-system/src/main/java/org/nl/config/mqtt2/msg/MsgWorker.java

@@ -0,0 +1,27 @@
+package org.nl.config.mqtt2.msg;
+
+/*
+ * @author ZZQ
+ * @Date 2024/1/31 15:29
+ * 消息处理线程：通过线程池处理：如果需要用到具体topic的handler：可以传入
+ */
+public class MsgWorker implements Runnable{
+    /**
+     * topic
+     */
+    public String topic;
+    /**
+     * 消息体
+     */
+    private String body;
+
+    @Override
+    public void run() {
+        System.out.println("接收到消息"+topic+"-"+body);
+    }
+
+    public MsgWorker(String topic, String body) {
+        this.topic = topic;
+        this.body = body;
+    }
+}

wcs/nladmin-system/src/main/java/org/nl/modules/loki/service/impl/LokiServiceImpl.java

@@ -18,10 +18,8 @@ import org.springframework.stereotype.Service;
 @RequiredArgsConstructor
 public class LokiServiceImpl implements LokiService {
 
-    @Value("${loki.url}")
     private String lokiUrl;
 
-    @Value("${loki.systemName}")
     private String systemName;
 
     @Override

wcs/nladmin-system/src/main/java/org/nl/modules/lucene/service/impl/LuceneServiceImpl.java

@@ -20,11 +20,6 @@ import java.util.Map;
 @Slf4j
 public class LuceneServiceImpl implements LuceneService {
 
-    @Value("${loki.url}")
-    private String lokiUrl;
-
-    @Value("${loki.systemName}")
-    private String systemName;
 
     //日志索引目录
     @Value("${lucene.index.path}")
@@ -39,12 +34,12 @@ public class LuceneServiceImpl implements LuceneService {
     public JSONArray getLabelsValues() {
         JSONArray result = new JSONArray();
         // 获取所有标签
-        String labelString = HttpUtil.get(lokiUrl + "/labels", CharsetUtil.CHARSET_UTF_8);
+        String labelString = HttpUtil.get(luceneUrl + "/labels", CharsetUtil.CHARSET_UTF_8);
         JSONObject parse = (JSONObject) JSONObject.parse(labelString);
         JSONArray labels = parse.getJSONArray("data");
         for (int i=0; i<labels.size(); i++) {
             // 获取标签下的所有值
-            String valueString = HttpUtil.get(lokiUrl + "/label/" + labels.getString(i) + "/values", CharsetUtil.CHARSET_UTF_8);
+            String valueString = HttpUtil.get(luceneUrl + "/label/" + labels.getString(i) + "/values", CharsetUtil.CHARSET_UTF_8);
             JSONObject parse2 = (JSONObject) JSONObject.parse(valueString);
             JSONArray values = parse2.getJSONArray("data");
             JSONArray children = new JSONArray();

wcs/nladmin-system/src/main/resources/config/application-dev.yml

@@ -161,7 +161,3 @@ sa-token:
   jwt-secret-key: opsjajisdnnca0sdkksdfaaasdfwwq
   # token 前缀
   token-prefix: Bearer
-
-loki:
-  url: http://127.0.0.1:3100/loki/api/v1
-  systemName: acs

wcs/nladmin-system/src/main/resources/config/application-dev3.yml

@@ -0,0 +1,176 @@
+server:
+  port: 8010
+  tomcat:
+    accept-count: 1000
+    max-connections: 10000
+    max-threads: 800
+    min-spare-threads: 100
+  shutdown: graceful
+#配置数据源
+spring:
+  lifecycle:
+    timeout-per-shutdown-phase: 60s
+  datasource:
+    druid:
+      db-type: com.alibaba.druid.pool.DruidDataSource
+      driverClassName: net.sf.log4jdbc.sql.jdbcapi.DriverSpy
+      #      url: jdbc:log4jdbc:mysql://${DB_HOST:10.1.3.91}:${DB_PORT:3306}/${DB_NAME:acs}?serverTimezone=Asia/Shanghai&characterEncoding=utf8&useSSL=false&useOldAliasMetadataBehavior=true
+      #     url: jdbc:log4jdbc:mysql://${DB_HOST:192.168.81.252}:${DB_PORT:3306}/${DB_NAME:hl_one_acs}?serverTimezone=Asia/Shanghai&characterEncoding=utf8&useSSL=false&useOldAliasMetadataBehavior=true
+      url: jdbc:log4jdbc:mysql://${DB_HOST:127.0.0.1}:${DB_PORT:3306}/${DB_NAME:hl_one_acs}?serverTimezone=Asia/Shanghai&characterEncoding=utf8&useSSL=false&useOldAliasMetadataBehavior=true&allowPublicKeyRetrieval=true
+      username: ${DB_USER:root}
+      #      password: ${DB_PWD:P@ssw0rd}
+      #     password: ${DB_PWD:Root.123456}
+      password: ${DB_PWD:942464Yy}
+
+      # 初始连接数
+      initial-size: 5
+      # 最小连接数
+      min-idle: 15
+      # 最大连接数
+      max-active: 30
+      # 超时时间(以秒数为单位)
+      remove-abandoned-timeout: 180
+      # 获取连接超时时间
+      max-wait: 3000
+      # 连接有效性检测时间
+      time-between-eviction-runs-millis: 60000
+      # 连接在池中最小生存的时间
+      min-evictable-idle-time-millis: 300000
+      # 连接在池中最大生存的时间
+      max-evictable-idle-time-millis: 900000
+      # 指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除
+      test-while-idle: true
+      # 指明是否在从池中取出连接前进行检验,如果检验失败, 则从池中去除连接并尝试取出另一个
+      test-on-borrow: true
+      # 是否在归还到池中前进行检验
+      test-on-return: false
+      # 检测连接是否有效
+      validation-query: select 1
+      # 配置监控统计
+      webStatFilter:
+        enabled: true
+      stat-view-servlet:
+        enabled: true
+        url-pattern: /druid/*
+        reset-enable: false
+      filter:
+        stat:
+          enabled: true
+          # 记录慢SQL
+          log-slow-sql: true
+          slow-sql-millis: 1000
+          merge-sql: true
+        wall:
+          config:
+            multi-statement-allow: true
+  redis:
+    #数据库索引
+    database: ${REDIS_DB:15}
+    host: ${REDIS_HOST:127.0.0.1}
+    port: ${REDIS_PORT:6379}
+    password: ${REDIS_PWD:}
+  mqtt:
+    username: a
+    password: a
+    url: tcp://10.211.55.3:1884
+    clientId: mqttdemo001
+    topics:
+      - test01
+      - test02
+    timeout: 10
+    keepalive: 100
+# 登录相关配置
+login:
+  # 登录缓存
+  cache-enable: true
+  #  是否限制单用户登录
+  single-login: false
+  #  验证码
+  login-code:
+    #  验证码类型配置 查看 LoginProperties 类
+    code-type: arithmetic
+    #  登录图形验证码有效时间/分钟
+    expiration: 2
+    #  验证码高度
+    width: 111
+    #  验证码宽度
+    heigth: 36
+    # 内容长度
+    length: 2
+    # 字体名称，为空则使用默认字体
+    font-name:
+    # 字体大小
+    font-size: 25
+
+#jwt
+jwt:
+  header: Authorization
+  # 令牌前缀
+  token-start-with: Bearer
+  # 必须使用最少88位的Base64对该令牌进行编码
+  base64-secret: ZmQ0ZGI5NjQ0MDQwY2I4MjMxY2Y3ZmI3MjdhN2ZmMjNhODViOTg1ZGE0NTBjMGM4NDA5NzYxMjdjOWMwYWRmZTBlZjlhNGY3ZTg4Y2U3YTE1ODVkZDU5Y2Y3OGYwZWE1NzUzNWQ2YjFjZDc0NGMxZWU2MmQ3MjY1NzJmNTE0MzI=
+  # 令牌过期时间 此处单位/毫秒 ，默认4小时，可在此网站生成 https://www.convertworld.com/zh-hans/time/milliseconds.html
+  token-validity-in-seconds: 14400000
+  # 在线用户key
+  online-key: online-token-
+  # 验证码
+  code-key: code-key-
+  # token 续期检查时间范围（默认30分钟，单位毫秒），在token即将过期的一段时间内用户操作了，则给用户的token续期
+  detect: 1800000
+  # 续期时间范围，默认1小时，单位毫秒
+  renew: 3600000
+
+#是否允许生成代码，生产环境设置为false
+generator:
+  enabled: true
+
+#是否开启 swagger-ui
+swagger:
+  enabled: true
+
+# IP 本地解析
+ip:
+  local-parsing: true
+
+# 文件存储路径
+file:
+  mac:
+    path: ~/file/
+    avatar: ~/avatar/
+  linux:
+    path: /home/eladmin/file/
+    avatar: /home/eladmin/avatar/
+  windows:
+    path: C:\eladmin\file\
+    avatar: C:\eladmin\avatar\
+  # 文件大小 /M
+  maxSize: 100
+  avatarMaxSize: 5
+logging:
+  file:
+    path: C:\logs\nlwms\
+  config: classpath:logback-spring.xml
+lucene:
+  index:
+    path: C:\logs\index
+
+# Sa-Token配置
+sa-token:
+  # token 名称 (同时也是cookie名称)
+  token-name: Authorization
+  # token 有效期，单位s 默认30天, -1代表永不过期
+  timeout: 2592000
+  # token 临时有效期 (指定时间内无操作就视为token过期) 单位: 秒
+  activity-timeout: -1
+  # 是否允许同一账号并发登录 (为true时允许一起登录, 为false时新登录挤掉旧登录)
+  is-concurrent: true
+  # 在多人登录同一账号时，是否共用一个token (为true时所有登录共用一个token, 为false时每次登录新建一个token)
+  is-share: false
+  # token风格
+  token-style: random-128
+  # 是否输出操作日志
+  is-log: false
+  jwt-secret-key: opsjajisdnnca0sdkksdfaaasdfwwq
+  # token 前缀
+  token-prefix: Bearer
+

wcs/nladmin-system/src/main/resources/config/application-prod.yml

@@ -165,7 +165,3 @@ sa-token:
   jwt-secret-key: opsjajisdnnca0sdkksdfaaasdfwwq
   # token 前缀
   token-prefix: Bearer
-
-loki:
-  url: http://localhost:3100/loki/api/v1
-  systemName: acs

wcs/nladmin-system/src/main/resources/config/application-test.yml

@@ -157,7 +157,3 @@ sa-token:
   jwt-secret-key: opsjajisdnnca0sdkksdfaaasdfwwq
   # token 前缀
   token-prefix: Bearer
-
-loki:
-  url: http://localhost:3100/loki/api/v1
-  systemName: acs

wcs/nladmin-system/src/main/resources/config/application.yml

@@ -78,3 +78,6 @@ security:
     - /api/localStorage/pictures
     # 参数
     - /api/param/getValueByCode
+lucene:
+  index:
+    path: D:\lucene\index

wcs/nladmin-system/src/main/resources/logback-spring.xml

@@ -14,9 +14,7 @@ https://juejin.cn/post/6844903775631572999
     <property name="log.pattern"
               value="%black(%contextName-) %red(%d{yyyy-MM-dd HH:mm:ss.SSS}) %green([%thread]) %highlight(%-5level) %boldMagenta(%logger{36}) - %gray(%msg%n)"/>
     <springProperty scope="context" name="logPath" source="logging.file.path" defaultValue="logs"/>
-    <springProperty scope="context" name="lokiUrl" source="loki.url"/>
     <springProperty scope="context" name="systemName" source="loki.systemName"/>
-    <property name="LOKI_URL" value="${lokiUrl}"/>
     <property name="SYSTEM_NAME" value="${systemName}"/>
     <!--定义日志文件的存储地址 勿在 LogBack 的配置中使用相对路径-->
     <property name="LOG_HOME" value="${logPath}"/>
@@ -66,26 +64,44 @@ https://juejin.cn/post/6844903775631572999
         <queueSize>500</queueSize>
         <appender-ref ref="FILE"/>
     </appender>
-
+    <appender name="luceneAppender" class="org.nl.config.lucene.LuceneAppender" >
-    <!--添加loki-->
+        <properties>
-    <appender name="lokiAppender" class="com.github.loki4j.logback.Loki4jAppender">
+            <property>
-        <batchTimeoutMs>1000</batchTimeoutMs>
+                <name>system</name>
-        <http class="com.github.loki4j.logback.ApacheHttpSender">
+                <value>lms</value>
-            <url>${LOKI_URL}/push</url>
+            </property>
-        </http>
+            <property>
-        <format>
+                <name>logLevel</name>
-            <label>
+                <value>%level</value>
-                <pattern>
+            </property>
-                    system=${SYSTEM_NAME},level=%level,logType=%X{log_file_type:-logType},device=%X{device_code_log:-device}
+            <property>
-                </pattern>
+                <name>requestMethod</name>
-            </label>
+                <value>%X{requestMethod}</value>
-            <message>
+            </property>
-                <pattern>${log.pattern}</pattern>
+            <property>
-            </message>
+                <name>requestTime</name>
-            <sortByTime>true</sortByTime>
+                <value>%d{yyyy-MM-dd HH:mm:ss.SSS}</value>
-        </format>
+            </property>
+            <property>
+                <name>requestIp</name>
+                <value>%X{requestIp}</value>
+            </property>
+            <property>
+                <name>thread</name>
+                <value>%thread</value>
+            </property>
+            <property>
+                <name>logger</name>
+                <value>%logger</value>
+            </property>
+        </properties>
+    </appender>
+    <appender name="asyncLuceneAppender" class="org.nl.config.lucene.AsyncLuceneAppender">
+        <appender-ref ref="luceneAppender" />
+        <queueSize>512</queueSize>
     </appender>
 
+    <!--添加loki-->
     <appender name="ERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
         <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
             <level>error</level>
@@ -108,24 +124,24 @@ https://juejin.cn/post/6844903775631572999
 
 
     <!--开发环境:打印控制台-->
-    <springProfile name="dev">
+    <springProfile name="dev3">
         <root level="info">
             <appender-ref ref="CONSOLE"/>
-            <appender-ref ref="lokiAppender"/>
+
             <appender-ref ref="asyncFileAppender"/>
         </root>
         <!--logmanage -->
         <logger name="org.nl.acs.log.service.impl.DeviceExecuteLogServiceImpl" level="info" additivity="false">
             <appender-ref ref="CONSOLE"/>
-            <appender-ref ref="lokiAppender"/>
+
         </logger>
         <logger name="jdbc.resultsettable" level="ERROR" additivity="false">
             <appender-ref ref="CONSOLE"/>
-            <appender-ref ref="lokiAppender"/>
+
         </logger>
         <logger name="org.openscada.opc.lib.da.Server" level="ERROR" additivity="false">
             <appender-ref ref="CONSOLE"/>
-            <appender-ref ref="lokiAppender"/>
+
         </logger>
         <!--logmanage -->
 
@@ -150,11 +166,8 @@ https://juejin.cn/post/6844903775631572999
         </logger>
     </springProfile>
     <!--测试环境:打印控制台-->
-    <springProfile name="test">
+    <springProfile name="dev">
         <!-- 打印sql -->
-        <logger name="org.nl.start.Init" level="info" additivity="false">
-            <appender-ref ref="FILE"/>
-        </logger>
         <root level="info">
             <appender-ref ref="CONSOLE"/>
 
@@ -165,24 +178,22 @@ https://juejin.cn/post/6844903775631572999
     <springProfile name="prod">
         <root level="info">
             <appender-ref ref="ERROR"/>
-            <appender-ref ref="lokiAppender"/>
+
             <appender-ref ref="asyncFileAppender"/>
         </root>
         <!--logmanage -->
         <logger name="org.nl.acs.log.service.impl.DeviceExecuteLogServiceImpl" level="info" additivity="false">
-            <appender-ref ref="lokiAppender"/>
+
         </logger>
         <logger name="jdbc.resultsettable" level="ERROR" additivity="false">
             <appender-ref ref="CONSOLE"/>
-            <appender-ref ref="lokiAppender"/>
+
         </logger>
         <logger name="org.openscada.opc.lib.da.Server" level="ERROR" additivity="false">
             <appender-ref ref="CONSOLE"/>
-            <appender-ref ref="lokiAppender"/>
+
         </logger>
         <!--logmanage -->
-
-
         <logger name="jdbc.audit" level="ERROR" additivity="false">
             <appender-ref ref="CONSOLE"/>
         </logger>