把 search-service 中处理 NginxLog 相关的代码全部放到 log 包中

search/search-service/src/main/java/cn/reghao/tnb/search/app/controller/NginxLogController.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/controller/NginxLogController.java

@@ -1,13 +1,12 @@
-package cn.reghao.tnb.search.app.controller;
+package cn.reghao.tnb.search.app.log.controller;
 
 import cn.reghao.tnb.common.db.SelectOption;
 import cn.reghao.tnb.common.web.WebResult;
-import cn.reghao.tnb.search.app.log.NginxLogService;
+import cn.reghao.tnb.search.app.log.service.NginxLogService;
 import cn.reghao.tnb.search.app.log.model.dto.DateTimeRange;
 import cn.reghao.tnb.search.app.log.model.vo.ChartData;
 import cn.reghao.tnb.search.app.log.model.vo.ChartMap;
 import cn.reghao.tnb.search.app.log.model.vo.GroupCount;
-import cn.reghao.tnb.search.app.log.pc.LogConsumer;
 import io.swagger.v3.oas.annotations.Operation;
 import org.springframework.http.MediaType;
 import org.springframework.validation.annotation.Validated;
@@ -23,18 +22,9 @@ import java.util.List;
 @RequestMapping("/api/search1/nginx")
 public class NginxLogController {
     private final NginxLogService nginxLogService;
-    private final LogConsumer logConsumer;
 
-    public NginxLogController(NginxLogService nginxLogService, LogConsumer logConsumer) {
+    public NginxLogController(NginxLogService nginxLogService) {
         this.nginxLogService = nginxLogService;
-        this.logConsumer = logConsumer;
-    }
-
-    @Operation(summary = "", description = "N")
-    @GetMapping(value = "/task", produces = MediaType.APPLICATION_JSON_VALUE)
-    public String logTask() {
-        int total = logConsumer.getActiveTasks();
-        return WebResult.success(total);
     }
 
     @Operation(summary = "NginxLog 日期列表", description = "N")

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/ip/GeoIpTool.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/geoip/GeoIpTool.java

@@ -1,4 +1,4 @@
-package cn.reghao.tnb.search.app.log.ip;
+package cn.reghao.tnb.search.app.log.geoip;
 
 /**
  * @author reghao

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/ip/IPLocation.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/geoip/IPLocation.java

@@ -1,4 +1,4 @@
-package cn.reghao.tnb.search.app.log.ip;
+package cn.reghao.tnb.search.app.log.geoip;
 
 import java.io.File;
 import java.io.IOException;

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/ip/Location.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/geoip/Location.java

@@ -1,4 +1,4 @@
-package cn.reghao.tnb.search.app.log.ip;
+package cn.reghao.tnb.search.app.log.geoip;
 
 import lombok.Getter;
 

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/kafka/KafkaDemo.java

@@ -1,20 +0,0 @@
-package cn.reghao.tnb.search.app.log.kafka;
-
-/**
- * @author reghao
- * @date 2022-04-02 11:20:21
- */
-public class KafkaDemo {
-    public static void main(String[] args) throws InterruptedException {
-        /*KafkaSub kafkaSub = new KafkaSub();
-        kafkaSub.consume();
-
-        KafkaPub kafkaPub = new KafkaPub();
-        int i = 1;
-        while (!Thread.interrupted()) {
-            kafkaPub.produce(i);
-            Thread.sleep(1_000);
-            i++;
-        }*/
-    }
-}

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/kafka/KafkaPub.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/mq/KafkaPub.java

@@ -1,4 +1,4 @@
-package cn.reghao.tnb.search.app.log.kafka;
+package cn.reghao.tnb.search.app.log.mq;
 
 import cn.reghao.jutil.jdk.serializer.JsonConverter;
 import cn.reghao.jutil.jdk.web.log.NginxLog;

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/kafka/KafkaSub.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/mq/KafkaSub.java

@@ -1,11 +1,10 @@
-package cn.reghao.tnb.search.app.log.kafka;
+package cn.reghao.tnb.search.app.log.mq;
 
 import cn.reghao.jutil.jdk.serializer.JsonConverter;
 import cn.reghao.jutil.jdk.thread.ThreadPoolWrapper;
 import cn.reghao.jutil.jdk.web.log.NginxLog;
 import cn.reghao.tnb.search.app.config.AppProperties;
-import cn.reghao.tnb.search.app.log.NginxLogService;
-import jakarta.annotation.PostConstruct;
+import cn.reghao.tnb.search.app.log.service.NginxLogService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.CommitFailedException;
 import org.apache.kafka.clients.consumer.ConsumerRecord;

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/consumer/RabbitListeners.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/mq/RabbitListeners.java

@@ -1,4 +1,4 @@
-package cn.reghao.tnb.search.app.log.consumer;
+package cn.reghao.tnb.search.app.log.mq;
 
 import cn.reghao.jutil.jdk.web.log.AppLog;;
 import cn.reghao.jutil.jdk.web.log.GatewayLog;

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/pc/LogConsumer.java

@@ -1,108 +0,0 @@
-package cn.reghao.tnb.search.app.log.pc;
-
-import cn.reghao.jutil.jdk.web.log.NginxLog;
-import cn.reghao.tnb.search.app.log.NginxLogService;
-import jakarta.annotation.PostConstruct;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-import org.springframework.stereotype.Component;
-
-import java.util.concurrent.Executor;
-
-/**
- * @author reghao
- * @date 2025-12-31 14:25:59
- */
-@Slf4j
-@Component
-public class LogConsumer {
-    private final Executor taskExecutor;
-    private final Object monitor;
-    private final LogProducer logProducer;
-    private final NginxLogService nginxLogService;
-
-    public LogConsumer(@Qualifier("taskExecutor") Executor taskExecutor, LogProducer logProducer,
-                       NginxLogService nginxLogService) {
-        this.taskExecutor = taskExecutor;
-        this.monitor = logProducer.getMonitor();
-        this.logProducer = logProducer;
-        this.nginxLogService = nginxLogService;
-    }
-
-    @PostConstruct
-    public void run() {
-        taskExecutor.execute(new ConsumerThread());
-        log.info("NginxLog producer-consumer 模型启动...");
-    }
-
-    public int getActiveTasks() {
-        int queueSize = logProducer.size();
-        ThreadPoolTaskExecutor executor = (ThreadPoolTaskExecutor) taskExecutor;
-        int capacity = executor.getQueueCapacity();
-        int size = executor.getQueueSize();
-        int activeCount = executor.getActiveCount();
-        log.info("{} -> {} {} {}", queueSize, capacity, size, activeCount);
-        return activeCount;
-    }
-
-    class ConsumerThread implements Runnable {
-        @Override
-        public void run() {
-            ThreadPoolTaskExecutor executor = (ThreadPoolTaskExecutor) taskExecutor;
-            int capacity = executor.getQueueCapacity();
-            while (!Thread.interrupted()) {
-                try {
-                    if (executor.getQueueSize() < capacity) {
-                        dispatch();
-                    } else {
-                        log.info("当前有 {} 个活跃线程, 休眠 1s 等待线程池空闲...", executor.getActiveCount());
-                        Thread.sleep(1_000);
-                    }
-                } catch (Exception e) {
-                    e.printStackTrace();
-                }
-            }
-        }
-    }
-
-    private void dispatch() throws Exception {
-        synchronized (monitor) {
-            Object object = logProducer.get();
-            if (object != null) {
-                if (object instanceof NginxLog) {
-                    NginxLog nginxLog = (NginxLog) object;
-                    String timeStr = nginxLog.getTimeIso8601().replace("+08:00", "");
-                    nginxLog.setTimeIso8601(timeStr);
-
-                    LogTask logTask = new LogTask(nginxLogService, nginxLog);
-                    taskExecutor.execute(logTask);
-                } else {
-                    log.error("Object 类型未知");
-                }
-            } else {
-                //log.info("调用 monitor.wait() 等待 DataProducer 中有数据可用");
-                monitor.wait();
-            }
-        }
-    }
-
-    static class LogTask implements Runnable {
-        private final NginxLogService nginxLogService;
-        private final NginxLog nginxLog;
-
-        public LogTask(NginxLogService nginxLogService, NginxLog nginxLog) {
-            this.nginxLogService = nginxLogService;
-            this.nginxLog = nginxLog;
-        }
-
-        @Override
-        public void run() {
-            try {
-                nginxLogService.processNginxLog(nginxLog);
-            } catch (Exception e) {
-                log.error("LogTask failed with message: {}", e.getMessage());
-            }
-        }
-    }
-}

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/pc/LogProducer.java

@@ -1,41 +0,0 @@
-package cn.reghao.tnb.search.app.log.pc;
-
-import org.springframework.stereotype.Component;
-
-import java.util.concurrent.LinkedBlockingQueue;
-
-/**
- * @author reghao
- * @date 2025-12-31 14:25:53
- */
-@Component
-public class LogProducer {
-    private final Object monitor;
-    private final LinkedBlockingQueue<Object> dataQueue;
-
-    public LogProducer() {
-        this.monitor = new Object();
-        this.dataQueue = new LinkedBlockingQueue<>(10_000);
-    }
-
-    public Object getMonitor() {
-        return monitor;
-    }
-
-    public void put(Object object) throws InterruptedException {
-        dataQueue.put(object);
-        synchronized (monitor) {
-            // 通知 consumer 线程有数据可用
-            monitor.notify();
-        }
-    }
-
-    public Object get() throws InterruptedException {
-        return dataQueue.poll();
-        //return dataQueue.take();
-    }
-
-    public int size() {
-        return dataQueue.size();
-    }
-}

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/NginxLogDocument.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/service/NginxLogDocument.java

@@ -1,4 +1,4 @@
-package cn.reghao.tnb.search.app.log;
+package cn.reghao.tnb.search.app.log.service;
 
 import cn.reghao.jutil.jdk.string.SnowFlake;
 import cn.reghao.jutil.jdk.web.log.NginxLog;

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/NginxLogSearch.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/service/NginxLogSearch.java

@@ -1,4 +1,4 @@
-package cn.reghao.tnb.search.app.log;
+package cn.reghao.tnb.search.app.log.service;
 
 import cn.reghao.tnb.search.app.es.ElasticService;
 import cn.reghao.jutil.jdk.web.log.NginxLog;

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/NginxLogService.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/service/NginxLogService.java

@@ -1,4 +1,4 @@
-package cn.reghao.tnb.search.app.log;
+package cn.reghao.tnb.search.app.log.service;
 
 import cn.reghao.jutil.jdk.converter.DateTimeConverter;
 import cn.reghao.jutil.jdk.io.TextFile;
@@ -8,8 +8,8 @@ import cn.reghao.tnb.common.db.SelectOption;
 import cn.reghao.tnb.search.app.config.AppProperties;
 import cn.reghao.tnb.search.app.es.EsQuery;
 import cn.reghao.tnb.search.app.es.SearchService;
-import cn.reghao.tnb.search.app.log.ip.GeoIpTool;
-import cn.reghao.tnb.search.app.log.ip.Location;
+import cn.reghao.tnb.search.app.log.geoip.GeoIpTool;
+import cn.reghao.tnb.search.app.log.geoip.Location;
 import cn.reghao.tnb.search.app.log.model.dto.DateTimeRange;
 import cn.reghao.tnb.search.app.log.model.vo.ChartData;
 import cn.reghao.tnb.search.app.log.model.vo.ChartMap;

search/search-service/src/main/java/cn/reghao/tnb/search/app/log/NgxLogService.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/service/NgxLogService.java

@@ -1,4 +1,4 @@
-package cn.reghao.tnb.search.app.log;
+package cn.reghao.tnb.search.app.log.service;
 
 import cn.reghao.jutil.jdk.converter.DateTimeConverter;
 import cn.reghao.jutil.jdk.web.log.NginxLog;

search/search-service/src/main/java/cn/reghao/tnb/search/app/config/BeanConfig.java → search/search-service/src/main/java/cn/reghao/tnb/search/app/log/util/BeanConfig.java

@@ -1,6 +1,7 @@
-package cn.reghao.tnb.search.app.config;
+package cn.reghao.tnb.search.app.log.util;
 
-import cn.reghao.tnb.search.app.log.ip.GeoIpTool;
+import cn.reghao.tnb.search.app.config.AppProperties;
+import cn.reghao.tnb.search.app.log.geoip.GeoIpTool;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 

search/search-service/src/main/java/cn/reghao/tnb/search/app/ws/handler/LogHandler.java

@@ -6,9 +6,8 @@ import cn.reghao.jutil.jdk.serializer.JdkSerializer;
 import cn.reghao.jutil.jdk.serializer.JsonConverter;
 import cn.reghao.jutil.jdk.web.log.GatewayLog;
 import cn.reghao.jutil.jdk.web.log.NginxLog;
-import cn.reghao.tnb.search.app.log.kafka.KafkaPub;
+import cn.reghao.tnb.search.app.log.mq.KafkaPub;
 import cn.reghao.tnb.search.app.ws.config.WebSocketPath;
-import cn.reghao.tnb.search.app.log.pc.LogProducer;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.*;
@@ -29,11 +28,9 @@ public class LogHandler implements WebSocketHandler {
     private final Map<String, WebSocketSession> pullAccessMap = new ConcurrentHashMap<>();
     private final Map<String, WebSocketSession> pushSessionMap = new ConcurrentHashMap<>();
     private final Map<String, WebSocketSession> pullSessionMap = new ConcurrentHashMap<>();
-    private final LogProducer logProducer;
-    private KafkaPub kafkaPub;
+    private final KafkaPub kafkaPub;
 
-    public LogHandler(LogProducer logProducer, KafkaPub kafkaPub) {
-        this.logProducer = logProducer;
+    public LogHandler(KafkaPub kafkaPub) {
         this.kafkaPub = kafkaPub;
     }