多线程优化，小时均值修改

src/main/java/com/cn/esermis/config/RedisQueueService.java

@@ -19,7 +19,13 @@ public class RedisQueueService {
         listOps.rightPush(queueKey, message);
     }
 
+    public boolean isQueueEmpty(String queueKey) {
+        Long size = listOps.size(queueKey);
+        return size == 0;
+    }
+
     public Object receiveMessage(String queueKey) {
+
         return listOps.leftPop(queueKey);
     }
 

src/main/java/com/cn/esermis/constant/Constant.java

@@ -26,6 +26,7 @@ public class Constant {
 
         public static final  String CN = "CN"; //大气环境污染源
         public static final  String SERVER = "server"; //大气环境污染源
+        public static final  String RECEIVEREPEAT = "receiverepeat"; //大气环境污染源
 
         public static final List<String> COMMAND = new ArrayList<>(Arrays.asList(HOUR, MINUTE)); // 解析数据编码
 

src/main/java/com/cn/esermis/dpld/business/MonitorBusiness.java

@@ -6,6 +6,7 @@ import com.cn.esermis.coding.PollutionGas;
 import com.cn.esermis.constant.Constant;
 import com.cn.esermis.datawarn.common.ExceptionDataDispose;
 import com.cn.esermis.dpld.entity.*;
+import com.cn.esermis.dpld.mapper.AirMonitorHourRecordMapper;
 import com.cn.esermis.dpld.model.OutletAndFactorVo;
 import com.cn.esermis.dpld.service.*;
 import com.cn.esermis.model.CP;
@@ -18,8 +19,10 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
+import java.math.BigDecimal;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
@@ -44,6 +47,8 @@ public class MonitorBusiness {
     @Resource
     private IAirMonitorHourRecordService airMonitorHourRecordService;
     @Resource
+    private AirMonitorHourRecordMapper airMonitorHourRecordMapper;
+    @Resource
     private IPostMarkingService postMarkingService;
     @Resource
     private RedisTemplate<String, Object> redisTemplate;
@@ -104,13 +109,9 @@ public class MonitorBusiness {
                             throw new RuntimeException(e);
                         }
                         try {
+                            // 计算时间
                             airMonitorMinuteRecord.setIsExceed(exceptionDataDispose.ExceedData(factorModel.getMN(), x.getFactor(), x.getAvg()!=null?x.getAvg().doubleValue():0));
-                            airMonitorMinuteRecord.setIsLose(exceptionDataDispose.LoseData(x.getMax()!=null?x.getMax().doubleValue():0,
-                                    x.getMin()!=null?x.getMin().doubleValue():0,
-                                    x.getAvg()!=null?x.getAvg().doubleValue():0));
-                            if ((airMonitorMinuteRecord.getIsExceed()!=null && !airMonitorMinuteRecord.getIsExceed().equals("1") )||
-                                    (airMonitorMinuteRecord.getIsLose()!=null && !airMonitorMinuteRecord.getIsLose().equals("1") )||
-                                    (airMonitorMinuteRecord.getIsSix()!=null && !airMonitorMinuteRecord.getIsSix().equals("1"))){
+                            if ((airMonitorMinuteRecord.getIsExceed()!=null && !airMonitorMinuteRecord.getIsExceed().equals("1"))){
                                 saveExcepLog(x, airMonitorMinuteRecord);
                             }
                         } catch (Exception e) {
@@ -121,52 +122,6 @@ public class MonitorBusiness {
                     });
                     airMonitorMinuteRecordService.saveBatch(listData);
                 }
-//                else if (factorModel.getCN().equals(Constant.REAL)) {
-//                    // 分钟数据入库
-//                    List<AirMonitorMinuteRecord> listData = new ArrayList<>();
-//                    cp.forEach(x -> {
-//                        AirMonitorMinuteRecord airMonitorMinuteRecord = new AirMonitorMinuteRecord();
-//                        airMonitorMinuteRecord.setQn(factorModel.getQN());
-//                        airMonitorMinuteRecord.setSt(factorModel.getST());
-//                        airMonitorMinuteRecord.setCn(factorModel.getCN());
-//                        airMonitorMinuteRecord.setPw(factorModel.getPW());
-//                        airMonitorMinuteRecord.setMn(factorModel.getMN());
-//                        airMonitorMinuteRecord.setFlag(factorModel.getFlag());
-//                        airMonitorMinuteRecord.setDataTime(factorModel.getDataTime());
-//                        airMonitorMinuteRecord.setFactor(x.getFactor());
-//                        airMonitorMinuteRecord.setMin(x.getMin());
-//                        airMonitorMinuteRecord.setMax(x.getMax());
-//                        airMonitorMinuteRecord.setAvg(x.getAvg());
-//                        airMonitorMinuteRecord.setCou(x.getCou());
-//                        airMonitorMinuteRecord.setFactorFlag(x.getFlag());
-//                        airMonitorMinuteRecord.setZsAvg(x.getZsAvg());
-//                        airMonitorMinuteRecord.setZsMax(x.getZsMax());
-//                        airMonitorMinuteRecord.setZsMin(x.getZsMin());
-//                        airMonitorMinuteRecord.setRtd(x.getRtd());
-//                        try {
-//                            airMonitorMinuteRecord.setWorkCondition(sendMessage(factorModel.getDataTime(),factorModel.getMN()));
-//                        } catch (ParseException e) {
-//                            throw new RuntimeException(e);
-//                        }
-//                        try {
-//                            airMonitorMinuteRecord.setIsExceed(exceptionDataDispose.ExceedData(factorModel.getMN(), x.getFactor(), x.getAvg()!=null?x.getAvg().doubleValue():0));
-//                            airMonitorMinuteRecord.setIsLose(exceptionDataDispose.LoseData(x.getMax()!=null?x.getMax().doubleValue():0,
-//                                    x.getMin()!=null?x.getMin().doubleValue():0,
-//                                    x.getAvg()!=null?x.getAvg().doubleValue():0));
-//                            if ((airMonitorMinuteRecord.getIsExceed()!=null && !airMonitorMinuteRecord.getIsExceed().equals("Y") )||
-//                                    (airMonitorMinuteRecord.getIsLose()!=null && !airMonitorMinuteRecord.getIsLose().equals("Y") )||
-//                                    (airMonitorMinuteRecord.getIsSix()!=null && !airMonitorMinuteRecord.getIsSix().equals("Y"))){
-//                                saveExcepLog(x, airMonitorMinuteRecord);
-//                            }
-//                        } catch (Exception e) {
-//                            log.error("调用 ExceptionDataDispose2 方法异常{} 当前参数为:{}",e,airMonitorMinuteRecord);
-//                        }
-//                        listData.add(airMonitorMinuteRecord);
-//                    });
-//                    if (!listData.isEmpty()) {
-//                        airMonitorMinuteRecordService.saveBatch(listData);
-//                    }
-//                }
                 else if (factorModel.getCN().equals(Constant.HOUR)) {
                     // 小时数据入库
                     List<AirMonitorHourRecord> listData = new ArrayList<>();
@@ -200,7 +155,6 @@ public class MonitorBusiness {
                                     x.getMin()!=null?x.getMin().doubleValue():0,
                                     x.getAvg()!=null?x.getAvg().doubleValue():0));
                             airMonitorHourRecord.setIsSix(exceptionDataDispose.SixData(factorModel.getMN(), x.getFactor(),x.getAvg()!=null?x.getAvg().doubleValue():0, factorModel.getDataTime()));
-//                            if (!airMonitorHourRecord.getIsExceed().equals("Y") || !airMonitorHourRecord.getIsLose().equals("Y") || !airMonitorHourRecord.getIsSix().equals("Y")){
                             if ((airMonitorHourRecord.getIsExceed()!=null && !airMonitorHourRecord.getIsExceed().equals("1") )||
                                     (airMonitorHourRecord.getIsLose()!=null && !airMonitorHourRecord.getIsLose().equals("1") )||
                                     (airMonitorHourRecord.getIsSix()!=null && !airMonitorHourRecord.getIsSix().equals("1"))){
@@ -241,12 +195,8 @@ public class MonitorBusiness {
                         }
                         try {
                             waterMonitorMinuteRecord.setIsExceed(exceptionDataDispose.ExceedData(factorModel.getMN(), x.getFactor(), x.getAvg()!=null?x.getAvg().doubleValue():0));
-                            waterMonitorMinuteRecord.setIsLose(exceptionDataDispose.LoseData(x.getMax()!=null?x.getMax().doubleValue():0,
-                                    x.getMin()!=null?x.getMin().doubleValue():0,
-                                    x.getAvg()!=null?x.getAvg().doubleValue():0));
-                            if ((waterMonitorMinuteRecord.getIsExceed()!=null && !waterMonitorMinuteRecord.getIsExceed().equals("1") )||
-                                    (waterMonitorMinuteRecord.getIsLose()!=null && !waterMonitorMinuteRecord.getIsLose().equals("1") )||
-                                    (waterMonitorMinuteRecord.getIsSix()!=null && !waterMonitorMinuteRecord.getIsSix().equals("1"))){
+
+                            if ((waterMonitorMinuteRecord.getIsExceed()!=null && !waterMonitorMinuteRecord.getIsExceed().equals("1"))){
                                 saveExcepLog(x, waterMonitorMinuteRecord);
                             }
                         } catch (Exception e) {
@@ -256,50 +206,6 @@ public class MonitorBusiness {
                     });
                     waterMonitorMinuteRecordService.saveBatch(listData);
                 }
-//                else if (factorModel.getCN().equals(Constant.REAL)) {
-//                    // 分钟数据入库
-//                    List<WaterMonitorMinuteRecord> listData = new ArrayList<>();
-//                    cp.forEach(x -> {
-//                        WaterMonitorMinuteRecord waterMonitorMinuteRecord = new WaterMonitorMinuteRecord();
-//                        waterMonitorMinuteRecord.setQn(factorModel.getQN());
-//                        waterMonitorMinuteRecord.setSt(factorModel.getST());
-//                        waterMonitorMinuteRecord.setCn(factorModel.getCN());
-//                        waterMonitorMinuteRecord.setPw(factorModel.getPW());
-//                        waterMonitorMinuteRecord.setMn(factorModel.getMN());
-//                        waterMonitorMinuteRecord.setFlag(factorModel.getFlag());
-//                        waterMonitorMinuteRecord.setDataTime(factorModel.getDataTime());
-//                        waterMonitorMinuteRecord.setFactor(x.getFactor());
-//                        waterMonitorMinuteRecord.setMin(x.getMin());
-//                        waterMonitorMinuteRecord.setMax(x.getMax());
-//                        waterMonitorMinuteRecord.setAvg(x.getAvg());
-//                        waterMonitorMinuteRecord.setCou(x.getCou());
-//                        waterMonitorMinuteRecord.setFactorFlag(x.getFlag());
-//                        waterMonitorMinuteRecord.setRtd(x.getRtd());
-//                        try {
-//                            waterMonitorMinuteRecord.setWorkCondition(sendMessage(factorModel.getDataTime(),factorModel.getMN()));
-//                        } catch (ParseException e) {
-//                            throw new RuntimeException(e);
-//                        }
-//                        try {
-//                            waterMonitorMinuteRecord.setIsExceed(exceptionDataDispose.ExceedData(factorModel.getMN(), x.getFactor(), x.getAvg()!=null?x.getAvg().doubleValue():0));
-//                            waterMonitorMinuteRecord.setIsLose(exceptionDataDispose.LoseData(x.getMax()!=null?x.getMax().doubleValue():0,
-//                                    x.getMin()!=null?x.getMin().doubleValue():0,
-//                                    x.getAvg()!=null?x.getAvg().doubleValue():0));
-//                            if ((waterMonitorMinuteRecord.getIsExceed()!=null && !waterMonitorMinuteRecord.getIsExceed().equals("Y") )||
-//                                    (waterMonitorMinuteRecord.getIsLose()!=null && !waterMonitorMinuteRecord.getIsLose().equals("Y") )||
-//                                    (waterMonitorMinuteRecord.getIsSix()!=null && !waterMonitorMinuteRecord.getIsSix().equals("Y"))){
-//                                saveExcepLog(x, waterMonitorMinuteRecord);
-//                            }
-//                        } catch (Exception e) {
-//                            log.error("调用 ExceptionDataDispose5 方法异常{} 当前参数为:{}",e,waterMonitorMinuteRecord);
-//                        }
-//
-//                        listData.add(waterMonitorMinuteRecord);
-//                    });
-//                    if (!listData.isEmpty()) {
-//                        waterMonitorMinuteRecordService.saveBatch(listData);
-//                    }
-//                }
                 else if
                 (factorModel.getCN().equals(Constant.HOUR)) {
                     // 小时数据入库
@@ -352,7 +258,7 @@ public class MonitorBusiness {
 
     private void saveExcepLog(CP x, AirMonitorMinuteRecord record) {
         ExcepLog excepLog = new ExcepLog();
-        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d ->d.getFactor()!=null && d.getFactor().equals(x.getFactor())).findFirst();
+        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d ->d.getMnCode()!=null && d.getMnCode().equals(record.getMn()) &&  d.getFactor()!=null && d.getFactor().equals(x.getFactor())).findFirst();
         if (first.isPresent()) {
             excepLog.setOutletName(first.get().getOutletName());
             excepLog.setOutletCode(first.get().getOutletCode());
@@ -363,20 +269,11 @@ public class MonitorBusiness {
             excepLog.setLoseData(record.getIsLose());
             excepLogService.save(excepLog);
         }
-        if (record.getIsLose()!=null && !record.getIsLose().equals("1")){
-            addExcepLog(record.getFactor(), "1");
-        }
-        if (record.getIsSix()!=null && !record.getIsSix().equals("1")){
-            addExcepLog(record.getFactor(), "2");
-        }
-        if (record.getIsExceed()!=null && !record.getIsExceed().equals("1")){
-            addExcepLog(record.getFactor(), "3");
-        }
     }
 
     private void saveExcepLog(CP x, AirMonitorHourRecord record) {
         ExcepLog excepLog = new ExcepLog();
-        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d -> d.getFactor()!=null && d.getFactor().equals(x.getFactor())).findFirst();
+        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d -> d.getMnCode()!=null && d.getMnCode().equals(record.getMn()) &&  d.getFactor()!=null && d.getFactor().equals(x.getFactor())).findFirst();
         if (first.isPresent()) {
             excepLog.setOutletName(first.get().getOutletName());
             excepLog.setOutletCode(first.get().getOutletCode());
@@ -386,21 +283,37 @@ public class MonitorBusiness {
             excepLog.setSixLast(record.getIsSix());
             excepLog.setLoseData(record.getIsLose());
             excepLogService.save(excepLog);
+            if (record.getIsLose()!=null && !record.getIsLose().equals("1")){
+                addExcepLog(record.getFactor(), "1",record.getMn());
+            }
+            if (record.getIsSix()!=null && !record.getIsSix().equals("1")){
+                addExcepLog(record.getFactor(), "2",record.getMn());
+            }
+            // 小时均值超限
+            String range = first.get().getRange();
+            if (range!=null && !range.isEmpty()){
+                if (range.contains("-")){
+                    String[] split = range.split("-");
+                    if (split.length > 2
+                            && ((isNumeric(range)  && new BigDecimal(split[0]).compareTo(x.getAvg()) < 0)
+                            || (isNumeric(range) && new BigDecimal(split[2]).compareTo(x.getAvg()) < 0))) {
+                        addExcepLog(record.getFactor(), "3",record.getMn());
+                    }
+                }else{
+                    if (isNumeric(range) && new BigDecimal(range).compareTo(x.getAvg()) > 0){
+                        addExcepLog(record.getFactor(), "3",record.getMn());
+                    }
+                }
+            }
         }
-        if (record.getIsLose()!=null && !record.getIsLose().equals("1")){
-            addExcepLog(record.getFactor(), "1");
-        }
-        if (record.getIsSix()!=null && !record.getIsSix().equals("1")){
-            addExcepLog(record.getFactor(), "2");
-        }
-        if (record.getIsExceed()!=null && !record.getIsExceed().equals("1")){
-            addExcepLog(record.getFactor(), "3");
-        }
+
     }
 
     private void saveExcepLog(CP x, WaterMonitorMinuteRecord record) {
         ExcepLog excepLog = new ExcepLog();
-        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d ->d.getFactor()!=null &&  d.getFactor().equals(x.getFactor())).findFirst();
+        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d ->
+                d.getMnCode()!=null && d.getMnCode().equals(record.getMn()) &&
+                d.getFactor()!=null &&  d.getFactor().equals(x.getFactor())).findFirst();
         if (first.isPresent()) {
             excepLog.setOutletName(first.get().getOutletName());
             excepLog.setOutletCode(first.get().getOutletCode());
@@ -411,19 +324,10 @@ public class MonitorBusiness {
             excepLog.setLoseData(record.getIsLose());
             excepLogService.save(excepLog);
         }
-        if (record.getIsLose()!=null && !record.getIsLose().equals("1")){
-            addExcepLog(record.getFactor(), "1");
-        }
-        if (record.getIsSix()!=null && !record.getIsSix().equals("1")){
-            addExcepLog(record.getFactor(), "2");
-        }
-        if (record.getIsExceed()!=null && !record.getIsExceed().equals("1")){
-            addExcepLog(record.getFactor(), "3");
-        }
     }
     private void saveExcepLog(CP x, WaterMonitorHourRecord record) {
         ExcepLog excepLog = new ExcepLog();
-        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d -> d.getFactor()!=null && d.getFactor().equals(x.getFactor())).findFirst();
+        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d -> d.getMnCode()!=null && d.getMnCode().equals(record.getMn()) && d.getFactor()!=null && d.getFactor().equals(x.getFactor())).findFirst();
         if (first.isPresent()) {
             excepLog.setOutletName(first.get().getOutletName());
             excepLog.setOutletCode(first.get().getOutletCode());
@@ -433,16 +337,34 @@ public class MonitorBusiness {
             excepLog.setSixLast(record.getIsSix());
             excepLog.setLoseData(record.getIsLose());
             excepLogService.save(excepLog);
+            if (record.getIsLose()!=null && !record.getIsLose().equals("1")){
+                addExcepLog(record.getFactor(), "1",record.getMn());
+            }
+            if (record.getIsSix()!=null && !record.getIsSix().equals("1")){
+                addExcepLog(record.getFactor(), "2",record.getMn());
+            }
+            // 小时均值超限
+            String range = first.get().getRange();
+            if (range!=null && !range.isEmpty()){
+                if (range.contains("-")){
+                    String[] split = range.split("-");
+                    if (split.length > 2
+                            && ((isNumeric(range)  && new BigDecimal(split[0]).compareTo(x.getAvg()) < 0)
+                            || (isNumeric(range) && new BigDecimal(split[2]).compareTo(x.getAvg()) < 0))) {
+                        addExcepLog(record.getFactor(), "3",record.getMn());
+                    }
+                }else{
+                    if (isNumeric(range) && new BigDecimal(range).compareTo(x.getAvg()) > 0){
+                        addExcepLog(record.getFactor(), "3",record.getMn());
+                    }
+                }
+            }
         }
-        if (record.getIsLose()!=null && !record.getIsLose().equals("1")){
-            addExcepLog(record.getFactor(), "1");
-        }
-        if (record.getIsSix()!=null && !record.getIsSix().equals("1")){
-            addExcepLog(record.getFactor(), "2");
-        }
-        if (record.getIsExceed()!=null && !record.getIsExceed().equals("1")){
-            addExcepLog(record.getFactor(), "3");
-        }
+
+    }
+
+    public boolean isNumeric(String str) {
+        return str.matches("\\d+(\\.\\d+)?"); // 正则表达式 \d 是数字字符，+ 表示一次或多次。
     }
 
     public Integer sendMessage(String time,String mn) throws ParseException {
@@ -488,9 +410,9 @@ public class MonitorBusiness {
     /**
      * 添加事后处置单信息
      */
-    public void addExcepLog(String factor,String dataType) {
+    public void addExcepLog(String factor,String dataType,String mn) {
         PostMarking postMarking = new PostMarking();
-        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d -> d.getFactor()!=null && d.getFactor().equals(factor)).findFirst();
+        Optional<OutletAndFactorVo> first = getloadData().stream().filter(d -> d.getMnCode()!=null && d.getMnCode().equals(mn) && d.getFactor()!=null && d.getFactor().equals(factor)).findFirst();
         if (first.isPresent()) {
             postMarking.setOutletName(first.get().getOutletName());
             postMarking.setOutletNumber(first.get().getOutletCode());
@@ -510,7 +432,7 @@ public class MonitorBusiness {
                     break;
                 case "3":
                     //瞬时值是否为超标数据
-                    postMarking.setManagementContent("数据异常：监测因子【"+pollutionGas.mean()+"("+pollutionGas.code()+")】瞬时值超标。");
+                    postMarking.setManagementContent("数据异常：监测因子【"+pollutionGas.mean()+"("+pollutionGas.code()+")】小时均值超限。");
                     break;
             }
             postMarkingService.save(postMarking);

src/main/java/com/cn/esermis/dpld/model/OutletAndFactorVo.java

@@ -19,10 +19,15 @@ public class OutletAndFactorVo {
 
 
     private String factor;
+
     private String mnCode;
+
     private String companyCode;
+
     private String departmentCode;
 
+    private String range;
+
 
 
 

src/main/java/com/cn/esermis/server/Consumer.java

@@ -9,9 +9,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 /**
  * 消费者类
@@ -29,34 +27,90 @@ public class Consumer {
     public void receive() throws InterruptedException {
         System.out.println("开始消费消息");
         // 开启线程池，如果有数据就启动线程消费
-        int numberOfThreads = 8;
         int tolot = 0;
-        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
 
-        while (true){
+        int corePoolSize = 10; // 核心线程数
+        int maximumPoolSize = 50; // 最大线程数
+        long keepAliveTime = 60L; // 空闲线程存活时间（秒）
+        TimeUnit unit = TimeUnit.SECONDS; // 时间单位
+        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(200); // 工作队列
+        ThreadFactory threadFactory = Executors.defaultThreadFactory(); // 线程工厂
+        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy(); // 拒绝策略
+        ThreadPoolExecutor executor = new ThreadPoolExecutor(
+                corePoolSize,
+                maximumPoolSize,
+                keepAliveTime,
+                unit,
+                workQueue,
+                threadFactory,
+                handler
+        );
+
+        while (true) {
+            //取出消息
+//            Object message = redisQueueService.receiveMessage(Constant.SERVER);
+            boolean message = redisQueueService.isQueueEmpty(Constant.SERVER);
+            if (message) {
+//                System.out.println("消息队列为空");
+                try {
+                    Thread.sleep(1000 * 5);
+                } catch (InterruptedException e) {
+                }
+            } else {
+                tolot++;
+                int finalTolot = tolot;
+                Object mge = redisQueueService.receiveMessage(Constant.SERVER);
+                try {
+                    System.out.println("消费消息------------：第（" + finalTolot + "）条");
+                    JSONObject jsonObject = (JSONObject) JSON.toJSON(mge);
+                    monitorBusiness.inboundOperations(jsonObject);
+                } catch (Exception e) {
+                    redisQueueService.sendMessage(Constant.RECEIVEREPEAT, mge);
+                    Thread.currentThread().interrupt();
+                }
+//                executor.submit(() -> {
+//                    Object mge = redisQueueService.receiveMessage(Constant.SERVER);
+//                    try {
+//                        System.out.println("消费消息------------：第（" + finalTolot + "）条");
+//                        JSONObject jsonObject = (JSONObject) JSON.toJSON(mge);
+//                        monitorBusiness.inboundOperations(jsonObject);
+//                    } catch (Exception e) {
+//                        redisQueueService.sendMessage(Constant.RECEIVEREPEAT, mge);
+//                        Thread.currentThread().interrupt();
+//                    }
+//                });
+            }
+        }
+
+    }
+
+
+    @Async
+    public void receiveRepeat() throws InterruptedException {
+        System.out.println("开始消费消息多线程失败的信息");
+        // 开启线程池，如果有数据就启动线程消费
+//        int numberOfThreads = 8;
+        int tolot = 0;
+        while (true) {
             //取出消息
-            Object message = redisQueueService.receiveMessage(Constant.SERVER);
-            if (message==null || message.equals("")){
+//            Object message = redisQueueService.receiveMessage(Constant.SERVER);
+            boolean message = redisQueueService.isQueueEmpty(Constant.RECEIVEREPEAT);
+            if (message) {
 //                System.out.println("消息队列为空");
                 try {
-                    Thread.sleep(1000*5);
+                    Thread.sleep(1000 * 5);
                 } catch (InterruptedException e) {
                 }
-            }else{
+            } else {
                 tolot++;
                 int finalTolot = tolot;
-                executorService.submit(new Runnable() {
-                    @Override
-                    public void run() {
-                        // 数据入库
-//                        System.out.println("消费消息------------："+message);
-                        System.out.println("消费消息------------：第（"+ finalTolot+"）条");
-                        JSONObject jsonObject = (JSONObject) JSON.toJSON(message);
-                        monitorBusiness.inboundOperations(jsonObject);
-                    }
-                });
+                Object mge = redisQueueService.receiveMessage(Constant.RECEIVEREPEAT);
+                System.out.println("消费消息------------：第（" + finalTolot + "）条");
+                JSONObject jsonObject = (JSONObject) JSON.toJSON(mge);
+                monitorBusiness.inboundOperations(jsonObject);
             }
         }
+
     }
 
 }

src/main/resources/static/mybatis/DataAcquisitionInstrumentMapper.xml

@@ -9,10 +9,11 @@
                a.outlet_code outletCode,
                c.factor `factor`,
                a.mn_code `mnCode`,
+               c.range_str `range`,
                a.company_code `companyCode`,
                a.department_code `departmentCode`
         from data_acquisition_instrument a
-                 left join analytical_instruments b on a.id = b.data_acquisition_instrument_id
+                 INNER join analytical_instruments b on a.id = b.data_acquisition_instrument_id
                  INNER join monitoring_factor c on b.id = c.analytical_instrument_id
     </select>
     <select id="getCompony" resultType="java.util.Map">