管道，流程，数据源切换，大屏，测量点位

src/main/java/com/sipai/config/RedissonConfig.java

@@ -10,6 +10,8 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.util.concurrent.TimeUnit;
+
 /**
  * @Author : YYJ
  * @CreateTime : 2021/9/14
@@ -33,6 +35,18 @@ public class RedissonConfig {
     @Value("${redis.password}")
     public String password;
     
+    @Value("${redis.timeout:10000}")
+    public int timeout;
+    
+    @Value("${redis.maxIdle:100}")
+    public int maxIdle;
+    
+    @Value("${redis.maxActive:300}")
+    public int maxActive;
+    
+    @Value("${redis.testOnBorrow:true}")
+    public boolean testOnBorrow;
+
     @Value("${cluster1.host.port:}")
     public String cluster1;
     @Value("${cluster2.host.port:}")
@@ -58,10 +72,33 @@ public class RedissonConfig {
             singleServerConfig.setAddress("redis://" + host + ":" + port);
             //设置连接密码
             singleServerConfig.setPassword(password);
+            // 设置超时时间（毫秒）
+            singleServerConfig.setTimeout(timeout);
+            // 设置连接池大小
+            singleServerConfig.setConnectionMinimumIdleSize(Math.min(10, maxIdle));
+            singleServerConfig.setConnectionPoolSize(maxActive);
+            // 设置空闲连接超时
+            singleServerConfig.setIdleConnectionTimeout(10000);
+            // 设置 ping 间隔，保持连接活跃
+            singleServerConfig.setPingConnectionInterval(10000);
+            // 设置失败重试次数
+            singleServerConfig.setRetryAttempts(3);
+            // 设置重试间隔
+            singleServerConfig.setRetryInterval(1500);
+            // 设置数据库索引
 //        singleServerConfig.setDatabase(Integer.valueOf(database));
         }else if(MODE_CLUSTER.equals(mode)){//集群
             conf.useClusterServers()
-                    .setScanInterval(2000);
+                    .setScanInterval(2000)
+                    .setTimeout(timeout)
+                    .setMasterConnectionMinimumIdleSize(Math.min(10, maxIdle))
+                    .setMasterConnectionPoolSize(maxActive)
+                    .setSlaveConnectionMinimumIdleSize(Math.min(10, maxIdle))
+                    .setSlaveConnectionPoolSize(maxActive)
+                    .setIdleConnectionTimeout(10000)
+                    .setPingConnectionInterval(10000)
+                    .setRetryAttempts(3)
+                    .setRetryInterval(1500);
             if(cluster1!=null && !cluster1.isEmpty()){
                 conf.useClusterServers().addNodeAddress("redis://" + cluster1);
             }
@@ -83,7 +120,7 @@ public class RedissonConfig {
 //                .setPassword(password);
         }
 
-        //使用json序列化方式
+        //使用 json 序列化方式
         Codec codec = new JsonJacksonCodec();
         conf.setCodec(codec);
         redisson = Redisson.create(conf);

src/main/java/com/sipai/controller/pipeline/PipelineDataController.java

@@ -47,6 +47,9 @@ public class PipelineDataController {
         if (request.getParameter("search_name") != null && !request.getParameter("search_name").isEmpty()) {
             wherestr += " and pipeline_name like '%" + request.getParameter("search_name") + "%'";
         }
+        if (request.getParameter("search_pipeline_area") != null && !request.getParameter("search_pipeline_area").isEmpty()) {
+            wherestr += " and pipeline_area = '" + request.getParameter("search_pipeline_area") + "'";
+        }
         PageHelper.startPage(page, rows);
         List<PipelineData> list = this.pipelineDataService.selectListByWhere(wherestr + orderstr);
         PageInfo<PipelineData> pInfo = new PageInfo<PipelineData>(list);

src/main/java/com/sipai/controller/work/MPointController.java

@@ -3,6 +3,7 @@ package com.sipai.controller.work;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
+import com.sipai.dao.repository.MPointRepo;
 import com.sipai.entity.base.Result;
 import com.sipai.entity.data.CurveMpoint;
 import com.sipai.entity.data.CurveRemark;
@@ -18,6 +19,7 @@ import com.sipai.entity.user.*;
 import com.sipai.entity.visual.JspElement;
 import com.sipai.entity.visual.VisualCacheData;
 import com.sipai.entity.work.MeasurePoint_DATA;
+import com.sipai.entity.work.MPointES;
 import com.sipai.entity.work.OverviewProduce;
 import com.sipai.entity.work.OverviewProduceBlot;
 import com.sipai.service.company.CompanyService;
@@ -63,6 +65,7 @@ import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.search.sort.SortBuilders;
 import org.elasticsearch.search.sort.SortOrder;
+import org.apache.log4j.Logger;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
@@ -91,6 +94,7 @@ import java.util.*;
 @Controller
 @RequestMapping("/work/mpoint")
 public class MPointController {
+    private static final Logger logger = Logger.getLogger(MPointController.class);
     @Resource
     private MPointService mPointService;
     @Resource
@@ -157,6 +161,8 @@ public class MPointController {
     private CompanyService companyService;
     @Resource
     private UserService userService;
+    @Autowired
+    private MPointRepo mPointRepo;
 
     /**
      * 从数据库查数据
@@ -414,29 +420,98 @@ public class MPointController {
                                   @RequestParam(value = "order", required = false) String order) {
         String companyId = request.getParameter("companyId");
         sort = "morder";
-        SortOrder sortOrder = null;
+        SortOrder sortOrder = SortOrder.ASC;
-//        if (order == null || order == "asc") {
-//            sortOrder = SortOrder.ASC;
-//        } else {
-//            sortOrder = SortOrder.DESC;
-//        }
-        sortOrder = SortOrder.ASC;
-        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
-        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
 
-        BoolQueryBuilder childBoolQueryBuilder = QueryBuilders.boolQuery();
+        // 获取查询参数
         String searchName = request.getParameter("search_name");
         String searchCode = request.getParameter("search_code");
         String pSectionId = request.getParameter("pSectionId");
         String type = request.getParameter("type");
         String signaltype = request.getParameter("signaltype");
+        String unitId = request.getParameter("unitId");
+
+        // 比对ES全量数据总数和数据库全量数据总数
+        long dbCount = this.mPointService.getDbTotalCount();
+        long esCount = this.mPointService.getEsTotalCount();
+        logger.info("getlistES数据比对 -> DB总数:" + dbCount + ", ES总数:" + esCount);
+
+        String result;
+        if (dbCount != esCount) {
+            // 数据不一致，将数据库全量数据更新到ES
+            logger.info("ES与DB数据总数不一致，执行全量同步并返回数据库结果");
+            this.mPointService.syncAllDbDataToEs();
+
+            // 构建数据库查询条件，返回数据库结果至前端
+            String wherestr = " where 1=1 ";
+            if (searchName != null && !searchName.isEmpty()) {
+                wherestr += " and (parmname like '%" + searchName + "%' or mpointcode like '%" + searchName + "%')";
+            }
+            if (searchCode != null && !searchCode.isEmpty()) {
+                wherestr += " and mpointcode like '%" + searchCode + "%' ";
+            }
+            if (type != null && !type.isEmpty() && !type.equals("all")) {
+                wherestr += " and source_type = '" + type + "' ";
+            }
+            if (pSectionId != null && !pSectionId.isEmpty()) {
+                wherestr += " and processSectionCode = '" + pSectionId + "' ";
+            }
+            if (signaltype != null && !signaltype.isEmpty() && !signaltype.equals("-1") && !signaltype.equals("all")) {
+                wherestr += " and SignalType = '" + signaltype + "' ";
+            }
+            // 设备id
+            if (request.getParameter("pid") != null && !request.getParameter("pid").isEmpty()) {
+                wherestr += " and equipmentId = '" + request.getParameter("pid") + "' ";
+            }
+            // 设备编号
+            if (request.getParameter("equipmentcardId") != null && !request.getParameter("equipmentcardId").isEmpty()) {
+                EquipmentCard equipmentCard = equipmentCardService.getEquipmentByEquipmentCardId(request.getParameter("equipmentcardId"));
+                if (equipmentCard != null) {
+                    wherestr += " and equipmentId = '" + equipmentCard.getId() + "' ";
+                }
+            }
+            // unitId -> bizid 过滤
+            if (unitId != null && !unitId.isEmpty()) {
+                List<Unit> units = this.unitService.getChildrenCBWByUnitId(unitId);
+                Unit unit = this.unitService.getUnitById(unitId);
+                if (unit != null) {
+                    units.add(unit);
+                }
+                if (units.size() > 0) {
+                    String bizids = "";
+                    for (int u = 0; u < units.size(); u++) {
+                        if (!bizids.isEmpty()) {
+                            bizids += "','";
+                        }
+                        bizids += units.get(u).getId();
+                    }
+                    wherestr += " and bizid in ('" + bizids + "') ";
+                }
+            }
+
+            String orderstr = " order by " + sort + " asc ";
+            PageHelper.startPage(page, rows);
+            List<MPoint> list = this.mPointService.selectListByWhere(companyId, wherestr + orderstr);
+            // 查询工艺段
+            for (MPoint mp : list) {
+                List<ProcessSection> processSections = processSectionService.selectSimpleListByWhere("where code = '" + mp.getProcesssectioncode() + "'");
+                if (processSections != null && processSections.size() > 0) {
+                    mp.setProcessSection(processSections.get(0));
+                }
+            }
+            PageInfo<MPoint> pi = new PageInfo<MPoint>(list);
+            JSONArray json = JSONArray.fromObject(list);
+            result = "{\"total\":" + pi.getTotal() + ",\"rows\":" + json + "}";
+        } else {
+            // 数据一致，直接返回ES数据至前端
+            NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
+            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
+            BoolQueryBuilder childBoolQueryBuilder = QueryBuilders.boolQuery();
 
             if (searchName != null && !searchName.isEmpty()) {
                 childBoolQueryBuilder.must(QueryBuilders.wildcardQuery("parmname.keyword", "*" + searchName + "*"));
             }
             if (searchCode != null && !searchCode.isEmpty()) {
                 childBoolQueryBuilder.must(QueryBuilders.wildcardQuery("mpointcode.keyword", "*" + searchCode + "*"));
-//            childBoolQueryBuilder.must(QueryBuilders.matchPhrasePrefixQuery("mpointcode", searchCode));
             }
             if (type != null && !type.isEmpty() && !type.equals("all")) {
                 childBoolQueryBuilder.must(QueryBuilders.matchPhraseQuery("source_type", type));
@@ -448,12 +523,12 @@ public class MPointController {
                 childBoolQueryBuilder.must(QueryBuilders.matchPhraseQuery("signaltype", signaltype));
             }
 
-        //设备id
+            // 设备id
             if (request.getParameter("pid") != null && !request.getParameter("pid").isEmpty()) {
                 childBoolQueryBuilder.must(QueryBuilders.matchPhraseQuery("equipmentid", request.getParameter("pid")));
             }
 
-        //设备编号
+            // 设备编号
             if (request.getParameter("equipmentcardId") != null && !request.getParameter("equipmentcardId").isEmpty()) {
                 EquipmentCard equipmentCard = equipmentCardService.getEquipmentByEquipmentCardId(request.getParameter("equipmentcardId"));
                 if (equipmentCard != null) {
@@ -461,31 +536,18 @@ public class MPointController {
                 }
             }
 
-//        if (request.getParameter("signaltag") != null && !request.getParameter("signaltag").isEmpty()) {
-//            String signalTag = request.getParameter("signaltag").replace(",", " ");
-//            childBoolQueryBuilder.must(QueryBuilders.matchQuery("signaltag", signalTag));
-//        }
-
-//        childBoolQueryBuilder.must(QueryBuilders.matchQuery("bizid", companyId));
-
-        String unitId = request.getParameter("unitId");
-
             if (unitId != null && !unitId.isEmpty()) {
                 List<Unit> units = this.unitService.getChildrenCBWByUnitId(unitId);
                 Unit unit = this.unitService.getUnitById(unitId);
                 if (unit != null) {
                     units.add(unit);
                 }
-            //二次查询
 
                 String bizids = "";
-
-            //根据搜索关键字如果存在对应组织id，并且在搜索的bizid范围内，则将关键字搜索优先搜索
                 if (searchName != null && !searchName.isEmpty()) {
                     int lastMatchNum = 1;
                     String keyword = "";
                     char[] sNchars = searchName.toCharArray();
-//                logger.info("units--"+JSONArray.fromObject(units).toString());
                     for (int u = 0; u < units.size(); u++) {
                         int matchNum = 0;
                         for (char item : sNchars) {
@@ -493,56 +555,31 @@ public class MPointController {
                                 matchNum++;
                             }
                         }
-//                    logger.info("unit-match--"+u+"--"+matchNum);
                         if (matchNum > lastMatchNum) {
                             keyword = units.get(u).getId();
                         }
-
                     }
                     if (!keyword.isEmpty()) {
                         units = this.getParent(keyword, units);
-                    //关键字提前，优先搜索
                         bizids = keyword;
                     }
-//                logger.info("units---match--"+JSONArray.fromObject(units).toString());
                 }
                 for (int u = 0; u < units.size(); u++) {
-//                if(!CommString.UNIT_TYPE_DEPT.equals(units.get(u).getType()) && !CommString.UNIT_TYPE_USER.equals(units.get(u).getType()) ){
                     if (!bizids.isEmpty()) {
                         bizids += " ";
                     }
                     bizids += units.get(u).getId();
-//                }
                 }
-//            System.out.println("测量点搜索查询bizId--" + bizids);
                 childBoolQueryBuilder.must(QueryBuilders.matchQuery("bizid", bizids));
             }
 
-
-//        System.out.println("---查询条件--" + childBoolQueryBuilder.toString());
             boolQueryBuilder.should(childBoolQueryBuilder);
             nativeSearchQueryBuilder.withSort(SortBuilders.fieldSort(sort).order(sortOrder));
             nativeSearchQueryBuilder.withQuery(boolQueryBuilder);
-
-//		Page<MPoint> list= this.mPointService.selectListByES(nativeSearchQueryBuilder);
-
-//		int total = list.size();
             nativeSearchQueryBuilder.withPageable(PageRequest.of(page - 1, rows));
             Page<MPoint> list = this.mPointService.selectListByES(nativeSearchQueryBuilder);
 
-
+            // 查询工艺段
-        /*for (MPoint mp : list) {
-            Unit unit = this.unitService.getUnitById(mp.getBizid());
-            if (unit != null) {
-                String unitName = unit.getSname();
-                if (unitName == null || unitName.isEmpty()) {
-                    unitName = unit.getName();
-                }
-                mp.setBizname(unitName);
-            }
-        }*/
-
-        //查询工艺段
             for (MPoint mp : list) {
                 List<ProcessSection> processSections = processSectionService.selectSimpleListByWhere("where code = '" + mp.getProcesssectioncode() + "'");
                 if (processSections != null && processSections.size() > 0) {
@@ -551,9 +588,9 @@ public class MPointController {
             }
 
             JSONArray json = JSONArray.fromObject(list.getContent());
+            result = "{\"total\":" + list.getTotalElements() + ",\"rows\":" + json + "}";
+        }
 
-
-        String result = "{\"total\":" + list.getTotalElements() + ",\"rows\":" + json + "}";
         model.addAttribute("result", result);
         return new ModelAndView("result");
     }

src/main/java/com/sipai/dao/scada/MPointDao.java

@@ -40,6 +40,10 @@ public class MPointDao  extends CommDaoImpl<MPoint>{
 		return list;
 	}
 	
+	public long selectCount() {
+		return getSqlSession().selectOne("scada.MPointMapper.selectCount");
+	}
+
 	public MPoint selectByMpointCode(String mpointcode){
 		MPoint mPoint = this.getSqlSession().selectOne("scada.MPointMapper.selectByMpointCode", mpointcode);
 		return mPoint;

src/main/java/com/sipai/entity/alarm/PipelineAreaEnum.java

@@ -0,0 +1,41 @@
+package com.sipai.entity.alarm;
+
+public enum PipelineAreaEnum {
+
+    PipelineArea_Out_Side("OUT_SIDE", "场外管道"),
+    PipelineArea_In_Side("In_SIDE", "场内管道");
+
+    private String code;
+    private String description;
+
+    PipelineAreaEnum(String code, String description) {
+        this.code = code;
+        this.description = description;
+    }
+
+    public String getCode() {
+        return code;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    public static PipelineAreaEnum getByCode(String code) {
+        for (PipelineAreaEnum pipelineEnum : PipelineAreaEnum.values()) {
+            if (pipelineEnum.getCode().equals(code)) {
+                return pipelineEnum;
+            }
+        }
+        return null;
+    }
+
+    public static PipelineAreaEnum getByName(String name) {
+        for (PipelineAreaEnum pipelineEnum : PipelineAreaEnum.values()) {
+            if (pipelineEnum.getDescription().equals(name)) {
+                return pipelineEnum;
+            }
+        }
+        return null;
+    }
+}

src/main/java/com/sipai/entity/pipeline/PipelineData.java

@@ -71,6 +71,12 @@ public class PipelineData extends SQLAdapter implements Serializable {
     @Column(name = "pipeline_invert_elevation_m", precision = 10, scale = 5)
     private BigDecimal pipelineInvertElevationM;
 
+    /**
+     * 管道场内场外标识（场外管道/场内管道）
+     */
+    @Column(name = "pipeline_area", length = 20)
+    private String pipelineArea;
+
     // 构造方法
     public PipelineData() {
     }
@@ -148,6 +154,14 @@ public class PipelineData extends SQLAdapter implements Serializable {
         this.pipelineInvertElevationM = pipelineInvertElevationM;
     }
 
+    public String getPipelineArea() {
+        return pipelineArea;
+    }
+
+    public void setPipelineArea(String pipelineArea) {
+        this.pipelineArea = pipelineArea;
+    }
+
     @Override
     public String toString() {
         return "PipelineData{" +
@@ -160,6 +174,7 @@ public class PipelineData extends SQLAdapter implements Serializable {
                 ", startGroundElevationM=" + startGroundElevationM +
                 ", endGroundElevationM=" + endGroundElevationM +
                 ", pipelineInvertElevationM=" + pipelineInvertElevationM +
+                ", pipelineArea='" + pipelineArea + '\'' +
                 '}';
     }
 }

src/main/java/com/sipai/mapper/pipeline/PipelineDataMapper.xml

@@ -11,10 +11,11 @@
     <result column="start_ground_elevation_m" property="startGroundElevationM" jdbcType="DECIMAL" />
     <result column="end_ground_elevation_m" property="endGroundElevationM" jdbcType="DECIMAL" />
     <result column="pipeline_invert_elevation_m" property="pipelineInvertElevationM" jdbcType="DECIMAL" />
+    <result column="pipeline_area" property="pipelineArea" jdbcType="VARCHAR" />
   </resultMap>
   <sql id="Base_Column_List" >
     id, pipeline_name, diameter_mm, length_m, start_burial_depth_m, end_burial_depth_m, 
-    start_ground_elevation_m, end_ground_elevation_m, pipeline_invert_elevation_m
+    start_ground_elevation_m, end_ground_elevation_m, pipeline_invert_elevation_m, pipeline_area
   </sql>
   <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.Long" >
     select 
@@ -30,11 +31,11 @@
     insert into tb_pipeline_data (pipeline_name, diameter_mm, 
       length_m, start_burial_depth_m, end_burial_depth_m, 
       start_ground_elevation_m, end_ground_elevation_m, 
-      pipeline_invert_elevation_m)
+      pipeline_invert_elevation_m, pipeline_area)
     values (#{pipelineName,jdbcType=VARCHAR}, #{diameterMm,jdbcType=DECIMAL}, 
       #{lengthM,jdbcType=DECIMAL}, #{startBurialDepthM,jdbcType=DECIMAL}, #{endBurialDepthM,jdbcType=DECIMAL}, 
       #{startGroundElevationM,jdbcType=DECIMAL}, #{endGroundElevationM,jdbcType=DECIMAL}, 
-      #{pipelineInvertElevationM,jdbcType=DECIMAL})
+      #{pipelineInvertElevationM,jdbcType=DECIMAL}, #{pipelineArea,jdbcType=VARCHAR})
   </insert>
   <insert id="insertSelective" parameterType="com.sipai.entity.pipeline.PipelineData" useGeneratedKeys="true" keyProperty="id" >
     insert into tb_pipeline_data
@@ -64,6 +65,9 @@
       <if test="pipelineInvertElevationM != null" >
         pipeline_invert_elevation_m,
       </if>
+      <if test="pipelineArea != null" >
+        pipeline_area,
+      </if>
     </trim>
     <trim prefix="values (" suffix=")" suffixOverrides="," >
 
@@ -91,6 +95,9 @@
       <if test="pipelineInvertElevationM != null" >
         #{pipelineInvertElevationM,jdbcType=DECIMAL},
       </if>
+      <if test="pipelineArea != null" >
+        #{pipelineArea,jdbcType=VARCHAR},
+      </if>
     </trim>
   </insert>
   <update id="updateByPrimaryKeySelective" parameterType="com.sipai.entity.pipeline.PipelineData" >
@@ -120,6 +127,9 @@
       <if test="pipelineInvertElevationM != null" >
         pipeline_invert_elevation_m = #{pipelineInvertElevationM,jdbcType=DECIMAL},
       </if>
+      <if test="pipelineArea != null" >
+        pipeline_area = #{pipelineArea,jdbcType=VARCHAR},
+      </if>
     </set>
     where id = #{id,jdbcType=BIGINT}
   </update>
@@ -132,7 +142,8 @@
       end_burial_depth_m = #{endBurialDepthM,jdbcType=DECIMAL},
       start_ground_elevation_m = #{startGroundElevationM,jdbcType=DECIMAL},
       end_ground_elevation_m = #{endGroundElevationM,jdbcType=DECIMAL},
-      pipeline_invert_elevation_m = #{pipelineInvertElevationM,jdbcType=DECIMAL}
+      pipeline_invert_elevation_m = #{pipelineInvertElevationM,jdbcType=DECIMAL},
+      pipeline_area = #{pipelineArea,jdbcType=VARCHAR}
     where id = #{id,jdbcType=BIGINT}
   </update>
   <select id="selectListByWhere" parameterType="java.lang.String" resultMap="BaseResultMap">

src/main/java/com/sipai/mapper/scada/MPointMapper.xml

@@ -764,6 +764,9 @@
             structureId        = #{structureId,jdbcType=VARCHAR}
         where ID = #{id,jdbcType=VARCHAR}
     </update>
+    <select id="selectCount" resultType="java.lang.Long">
+        select count(1) from tb_measurepoint
+    </select>
     <select id="selectListByWhere" parameterType="java.lang.String" resultMap="BaseResultMap">
         select
         <include refid="Base_Column_List"/>

src/main/java/com/sipai/netty/NettyClient.java

@@ -4,10 +4,14 @@ import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.concurrent.GenericFutureListener;
+
+import java.util.concurrent.TimeUnit;
 
 public class NettyClient extends Thread {
     private int port;
@@ -16,6 +20,9 @@ public class NettyClient extends Thread {
     //private String host = "132.120.132.74";//公司内网
     
     private Channel channel;
+    private EventLoopGroup clientWorker;
+    private volatile boolean running = true;
+    private Bootstrap bootstrap;
 
     public NettyClient(int port){
         this.port = port;
@@ -29,49 +36,39 @@ public class NettyClient extends Thread {
         /**
          * 负责装客户端的事件处理线程池
          */
-        EventLoopGroup clientWorker = new NioEventLoopGroup();
+        clientWorker = new NioEventLoopGroup();
         try {
             /**
              * netty客户端引导启动器
              */
-            Bootstrap bootstrap = new Bootstrap();
+            bootstrap = new Bootstrap();
             bootstrap
                     .group(clientWorker)//把事件处理线程池添加进启动引导器
                     .channel(NioSocketChannel.class)//设置通道的建立方式,这里采用Nio的通道方式来建立请求连接
-                    //.option(ChannelOption.SO_KEEPALIVE, true)
+                    .option(ChannelOption.SO_KEEPALIVE, true) // 启用心跳保活机制
+                    .option(ChannelOption.TCP_NODELAY, true) // 禁用Nagle算法，提高响应速度
+                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) // 连接超时10秒
                     .handler(new ChannelInitializer<SocketChannel>() {
                         @Override
                         protected void initChannel(SocketChannel socketChannel) throws Exception {
                             /**
                              * 此处添加客户端的通道处理器
                              */
-                            socketChannel.pipeline().addLast(new NettyClientHandler());
+                            socketChannel.pipeline().addLast(new NettyClientHandler(NettyClient.this));
                         }
                     });
-            /**
+            
-             * 客户端绑定端口并且开始发起连接请求
+            // 开始连接
-             */
+            connect();
-            ChannelFuture future = null;
+            
-            try {
-                future = bootstrap.connect(host, port).sync();
-            } catch (InterruptedException e) {
-                // TODO Auto-generated catch block
-                e.printStackTrace();
-            }
-            if (future.isSuccess()){
-                System.out.println("客户端连接服务器成功!");
-            }
-            /**
-             * 将通道设置好, 以便外面获取
-             */
-            this.channel = future.channel();
             /**
              * 关闭客户端
              */
             try {
-                future.channel().closeFuture().sync();
+                if (channel != null) {
+                    channel.closeFuture().sync();
+                }
             } catch (InterruptedException e) {
-                // TODO Auto-generated catch block
                 e.printStackTrace();
             }
             System.out.println("客户端即将关闭!");
@@ -84,6 +81,76 @@ public class NettyClient extends Thread {
         }
     }
 
+    /**
+     * 连接服务器
+     */
+    public void connect() {
+        if (!running) {
+            return;
+        }
+        System.out.println("正在连接服务器 " + host + ":" + port + " ...");
+        ChannelFuture future = null;
+        try {
+            future = bootstrap.connect(host, port);
+            future.addListener(new GenericFutureListener<ChannelFuture>() {
+                @Override
+                public void operationComplete(ChannelFuture future) throws Exception {
+                    if (future.isSuccess()) {
+                        System.out.println("客户端连接服务器成功!");
+                    } else {
+                        System.out.println("客户端连接服务器失败，5秒后重连...");
+                        future.channel().eventLoop().schedule(new Runnable() {
+                            @Override
+                            public void run() {
+                                connect();
+                            }
+                        }, 5, TimeUnit.SECONDS);
+                    }
+                }
+            });
+            future.sync();
+            this.channel = future.channel();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        } catch (Exception e) {
+            System.out.println("连接异常：" + e.getMessage() + "，5秒后重连...");
+            clientWorker.schedule(new Runnable() {
+                @Override
+                public void run() {
+                    connect();
+                }
+            }, 5, TimeUnit.SECONDS);
+        }
+    }
+
+    /**
+     * 连接断开时调用，进行重连
+     */
+    public void reconnect() {
+        if (running && clientWorker != null) {
+            System.out.println("连接断开，5秒后重连...");
+            clientWorker.schedule(new Runnable() {
+                @Override
+                public void run() {
+                    connect();
+                }
+            }, 5, TimeUnit.SECONDS);
+        }
+    }
+
+    /**
+     * 停止客户端
+     */
+    public void stopClient() {
+        running = false;
+        if (channel != null) {
+            channel.close();
+        }
+        if (clientWorker != null) {
+            clientWorker.shutdownGracefully();
+        }
+    }
+
     public Channel getChannel(){
         return this.channel;
     }

src/main/java/com/sipai/netty/NettyClientHandler.java

@@ -5,10 +5,18 @@ import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 
+import java.io.IOException;
 import java.nio.charset.Charset;
 import java.util.Date;
 
 public class NettyClientHandler extends ChannelInboundHandlerAdapter  {
+    
+    private NettyClient nettyClient;
+    
+    public NettyClientHandler(NettyClient nettyClient) {
+        this.nettyClient = nettyClient;
+    }
+    
     /**
      * 通道信息读取处理
      * @param ctx
@@ -25,6 +33,8 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter  {
              * 给服务端回复消息
              */
             ctx.writeAndFlush("客户端收到! 消息为: " + m.toString(Charset.defaultCharset()));
+        } catch (Exception e) {
+            System.err.println("读取消息异常: " + e.getMessage());
         } finally {
             m.release();
         }
@@ -42,15 +52,56 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter  {
          * 消息格式必须是ByteBuf才行!!!!!
          * 如果是其他格式服务端是接收不到的!!!!
          */
+        System.out.println("首次连接完成!");
         String helo = "你好呀!";
         ByteBuf byteBuf = Unpooled.wrappedBuffer(helo.getBytes());
         ctx.channel().writeAndFlush(byteBuf);
-        System.out.println("首次连接完成!");
     }
 
+    /**
+     * 当连接断开时调用
+     * @param ctx
+     * @throws Exception
+     */
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        System.out.println("连接已断开，准备重连...");
+        if (nettyClient != null) {
+            nettyClient.reconnect();
+        }
+    }
+
+    /**
+     * 异常处理
+     * @param ctx
+     * @param cause
+     */
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-        cause.printStackTrace();
+        // 处理远程主机强制关闭连接的异常
+        if (cause instanceof IOException) {
+            String message = cause.getMessage();
+            if (message != null && (message.contains("远程主机强迫关闭了一个现有的连接") 
+                    || message.contains("Connection reset")
+                    || message.contains("远程主机强迫关闭") 
+                    || message.contains("An existing connection was forcibly closed"))) {
+                System.out.println("[" + java.time.LocalDateTime.now() + "] 远程主机已关闭连接，等待重连...");
                 ctx.close();
+                // 触发重连
+                if (nettyClient != null) {
+                    nettyClient.reconnect();
+                }
+                return;
+            }
+        }
+        
+        // 其他异常只打印简要信息，不打印完整堆栈
+        System.err.println("[" + java.time.LocalDateTime.now() + "] 客户端异常：" + cause.getClass().getSimpleName() + " - " + cause.getMessage());
+        ctx.close();
+        
+        // 触发重连
+        if (nettyClient != null) {
+            nettyClient.reconnect();
+        }
     }
 }

src/main/java/com/sipai/netty/NettyServerHandler.java

@@ -15,6 +15,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.util.CharsetUtil;
 
+import java.io.IOException;
 import java.math.BigDecimal;
 import java.util.Map.Entry;
 
@@ -107,13 +108,36 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter  {
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-        super.exceptionCaught(ctx, cause);
+        // 处理远程主机强制关闭连接的异常
+        if (cause instanceof IOException) {
+            String message = cause.getMessage();
+            if (message != null && (message.contains("远程主机强迫关闭了一个现有的连接") 
+                    || message.contains("Connection reset")
+                    || message.contains("远程主机强迫关闭") 
+                    || message.contains("An existing connection was forcibly closed"))) {
+                System.out.println("[" + java.time.LocalDateTime.now() + "] 客户端 [" + ctx.channel().remoteAddress() + "] 已断开连接");
+                ctx.close();
+                return;
+            }
+        }
         /**
-         * 异常捕获
+         * 异常捕获 - 只打印简要信息
          */
+        System.err.println("[" + java.time.LocalDateTime.now() + "] 服务器处理异常：" + cause.getClass().getSimpleName() + " - " + cause.getMessage());
         cause.printStackTrace();
         ctx.close();
     }
+    
+    /**
+     * 当连接断开时调用
+     * @param ctx
+     * @throws Exception
+     */
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        System.out.println("[" + java.time.LocalDateTime.now() + "] 客户端 [" + ctx.channel().remoteAddress() + "] 连接断开");
+        super.channelInactive(ctx);
+    }
     public static boolean isNumber(String str){
         String reg = "^[0-9]+(.[0-9]+)?$";
         return str.matches(reg);

src/main/java/com/sipai/service/equipment/EquipmentCardService.java

@@ -340,9 +340,11 @@ public class EquipmentCardService implements CommService<EquipmentCard> {
 
             if (item.getEquipmentBelongId() != null && !item.getEquipmentBelongId().isEmpty()) {
                 EquipmentBelong equipmentBelong = this.equipmentBelongService.selectById(item.getEquipmentBelongId());
+                if (equipmentBelong != null) {
                     item.set_equipmentBelongName(equipmentBelong.getBelongName());
                 }
             }
+        }
         return equipmentCards;
     }
 
@@ -361,9 +363,9 @@ public class EquipmentCardService implements CommService<EquipmentCard> {
         for (EquipmentCard item : equipmentCards) {
             if (null != item.getId() && !item.getId().isEmpty()) {
                 EquipmentCardProp cardProp = this.propService.selectByEquipmentId(item.getId());
-                if (cardProp != null && cardProp.getKeynum() != null) {
+                if (cardProp != null) {
                     item.setEquipmentCardProp(cardProp);
-                    //查找该设备下的测量点，并将故障点为0的状态给equipmentcardprop的_state
+                    //查找该设备下的测量点，并将故障点为 0 的状态给 equipmentcardprop 的_state
                     List<MPoint> mPoints = this.mPointService.selectListByWhere(bizid, "where equipmentId = '" + item.getId() + "' and parmName like '%故障%' ");
         
                     if (mPoints != null && mPoints.size() > 0) {
@@ -459,7 +461,7 @@ public class EquipmentCardService implements CommService<EquipmentCard> {
                 if (null == item.getEquipmentClearTime()) {
                     item.setEquipmentClearTime(0.0);
                 }
-                if (cardProp.getRunTime() != null) {
+                if (cardProp != null && cardProp.getRunTime() != null) {
                     if (item.getEquipmentSetTime() < cardProp.getRunTime() - item.getEquipmentClearTime()) {
                         remindequipmentCards.add(item);
                     }

src/main/java/com/sipai/service/maintenance/impl/RepairServiceImpl.java

@@ -41,28 +41,33 @@ public class RepairServiceImpl implements RepairService{
 		Repair item=repairDao.selectByPrimaryKey(id);
 		//维修类型
 		EquipmentPlanType eqPlanType=this.equipmentPlanTypeService.selectById(item.getRepairType());
+		if(eqPlanType != null){
 			item.setRepairTypeName(eqPlanType.getName());
+		}
 		//接收人员们
-		String[] receiveUserids=item.getReceiveUserIds().split(",");
+		String receiveUserIds = item.getReceiveUserIds();
-		if(receiveUserids!=null&&receiveUserids.length>0){
+		if(receiveUserIds != null && !receiveUserIds.isEmpty()){
-			String receiveUsersName="";
+			String[] receiveUseridsArray = receiveUserIds.split(",");
-			for(int u=0;u<receiveUserids.length;u++){
+			if(receiveUseridsArray != null && receiveUseridsArray.length > 0){
-				User user=this.userService.getUserById(receiveUserids[u]);
+				String receiveUsersName = "";
-				if(user!=null){
+				for(int u = 0; u < receiveUseridsArray.length; u++){
-					receiveUsersName+=user.getCaption()+",";
+					User user = this.userService.getUserById(receiveUseridsArray[u]);
+					if(user != null){
+						receiveUsersName += user.getCaption() + ",";
 					}
 				}
-			if(receiveUsersName.length()>0){
+				if(receiveUsersName.length() > 0){
-				item.setReceiveUsersName(receiveUsersName.substring(0, receiveUsersName.length()-1));
+					item.setReceiveUsersName(receiveUsersName.substring(0, receiveUsersName.length() - 1));
+				}
 			}
 		}
 		//接收人员
-		User user=this.userService.getUserById(item.getReceiveUserId());
+		User user = this.userService.getUserById(item.getReceiveUserId());
-		if(user!=null){
+		if(user != null){
 			item.setReceiveUserName(user.getCaption());
 		}
 		//设备
-		EquipmentCard eCard=this.equipmentCardService.selectById(item.getEquipmentId());
+		EquipmentCard eCard = this.equipmentCardService.selectById(item.getEquipmentId());
 		item.setEquipmentCard(eCard);
 		
 		return item;
@@ -91,28 +96,33 @@ public class RepairServiceImpl implements RepairService{
 		for (Repair item : list) {
 			//维修类型
 			EquipmentPlanType eqPlanType=this.equipmentPlanTypeService.selectById(item.getRepairType());
+			if(eqPlanType != null){
 				item.setRepairTypeName(eqPlanType.getName());
+			}
 			//接收人员们
-			String[] receiveUserids=item.getReceiveUserIds().split(",");
+			String receiveUserIds = item.getReceiveUserIds();
-			if(receiveUserids!=null&&receiveUserids.length>0){
+			if(receiveUserIds != null && !receiveUserIds.isEmpty()){
-				String receiveUsersName="";
+				String[] receiveUseridsArray = receiveUserIds.split(",");
-				for(int u=0;u<receiveUserids.length;u++){
+				if(receiveUseridsArray != null && receiveUseridsArray.length > 0){
-					User user=this.userService.getUserById(receiveUserids[u]);
+					String receiveUsersName = "";
-					if(user!=null){
+					for(int u = 0; u < receiveUseridsArray.length; u++){
-						receiveUsersName+=user.getCaption()+",";
+						User user = this.userService.getUserById(receiveUseridsArray[u]);
+						if(user != null){
+							receiveUsersName += user.getCaption() + ",";
 						}
 					}
-				if(receiveUsersName.length()>0){
+					if(receiveUsersName.length() > 0){
-					item.setReceiveUsersName(receiveUsersName.substring(0, receiveUsersName.length()-1));
+						item.setReceiveUsersName(receiveUsersName.substring(0, receiveUsersName.length() - 1));
+					}
 				}
 			}
 			//接收人员
-			User user=this.userService.getUserById(item.getReceiveUserId());
+			User user = this.userService.getUserById(item.getReceiveUserId());
-			if(user!=null){
+			if(user != null){
 				item.setReceiveUserName(user.getCaption());
 			}
 			//设备
-			EquipmentCard eCard=this.equipmentCardService.selectById(item.getEquipmentId());
+			EquipmentCard eCard = this.equipmentCardService.selectById(item.getEquipmentId());
 			item.setEquipmentCard(eCard);
 			
 		}

src/main/java/com/sipai/service/scada/MPointService.java

@@ -696,6 +696,50 @@ public class MPointService {
         return mPage;
     }
 
+    /**
+     * 获取数据库中测量点总数
+     */
+    public long getDbTotalCount() {
+        return mPointDao.selectCount();
+    }
+
+    /**
+     * 获取ES中测量点总数
+     */
+    public long getEsTotalCount() {
+        return mPointRepo.count();
+    }
+
+    /**
+     * 将数据库全量数据同步到ES
+     */
+    public void syncAllDbDataToEs() {
+        logger.info("ES与数据库数据总数不一致，开始全量同步数据库数据到ES...");
+        MPoint query = new MPoint();
+        query.setWhere("where 1=1");
+        List<MPoint> allDbData = mPointDao.selectListByWhere(query);
+        if (allDbData != null && allDbData.size() > 0) {
+            List<MPointES> esList = new ArrayList<>();
+            for (MPoint mPoint : allDbData) {
+                try {
+                    MPointES mPointES = MPointES.format(mPoint);
+                    esList.add(mPointES);
+                } catch (Exception e) {
+                    logger.error("转换MPointES失败, id=" + mPoint.getId() + ", error=" + e.getMessage());
+                }
+            }
+            if (esList.size() > 0) {
+                try {
+                    mPointRepo.saveAll(esList);
+                    logger.info("全量同步完成，共同步 " + esList.size() + " 条数据到ES");
+                } catch (Exception e) {
+                    logger.error("批量保存ES失败: " + e.getMessage());
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
     @Transactional
     public int updateValue(String bizId, MPoint entity, MPointHistory mPointHistory) {
         try {

src/main/java/com/sipai/tools/DataSourceInterceptor.java

@@ -1,6 +1,7 @@
 package com.sipai.tools;
 
 import org.aspectj.lang.JoinPoint;
+import org.aspectj.lang.Signature;
 
 /*@Aspect	// for aop
 @Component // for auto scan
@@ -12,24 +13,20 @@ public class DataSourceInterceptor {
 
     //@Before("dataSourceScadaPointcut()")
     public void before(JoinPoint jp) {
-        Object[] argusObjects = jp.getArgs();
+//        Object[] argusObjects = jp.getArgs();
-
+        Signature signature = jp.getSignature();
         //若只有一个参数，则默认使用es
-        if (argusObjects == null || argusObjects.length < 2 || !(argusObjects[0] instanceof String)) {
+//        if (argusObjects == null || argusObjects.length < 2 || !(argusObjects[0] instanceof String)) {
-            return;
+//            return;
-        }
+//        }
-
+        String fullMethodPath = signature.toString();
-        Object argus = argusObjects[0];
+//        Object argus = argusObjects[0];
-        if (argus != null) {
+        if (fullMethodPath.contains(".scada.")) {
-			DataSourceHolder.setDataSources(DataSources.valueOf("SCADA_"+argus.toString()));
-//            DataSourceHolder.setDataSources(DataSources.valueOf("SCADA_HFCG"));
-        } else {
-//			DataSourceHolder.setDataSources(DataSources.valueOf("SCADA_YL"));
             DataSourceHolder.setDataSources(DataSources.valueOf("SCADA_0533JS"));
+        } else {
+            DataSourceHolder.setDataSources(DataSources.valueOf("MASTER"));
         }
 
-        //适应所有生产库存集中为一个数据库
-//		DataSourceHolder.setDataSources(DataSources.SCADA_0756ZH);
     }
 
     //@After("dataSourceScadaPointcut()")

src/main/webapp/WEB-INF/spring-mybatis-scada.xml

@@ -8,7 +8,7 @@
                         http://www.springframework.org/schema/context/spring-context-4.1.xsd
                         http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.1.xsd">
 
-    <bean id="SCADA_021HQWS"  class="com.alibaba.druid.pool.DruidDataSource" init-method="init"
+    <bean id="SCADA_0533JS"  class="com.alibaba.druid.pool.DruidDataSource" init-method="init"
           destroy-method="close">
         <property name="driverClassName" value="${driver}" />
         <property name="url" value="${scada-url}" />

src/main/webapp/WEB-INF/spring-mybatis.xml

@@ -175,7 +175,7 @@
             <map key-type="com.sipai.tools.DataSources">
                 <entry key="MASTER" value-ref="dataSourceMaster"/>
 <!--                <entry key="SCADA_HFCG" value-ref="SCADA_HFCG"/>-->
-                <entry key="SCADA_0533JS" value-ref="SCADA_021HQWS"/>
+                <entry key="SCADA_0533JS" value-ref="SCADA_0533JS"/>
                 <!-- 这里还可以加多个dataSource -->
             </map>
         </property>

src/main/webapp/jsp/pipeline/pipelineDataAdd.jsp

@@ -106,6 +106,16 @@
                             <input type="text" class="form-control" id="pipelineName" name="pipelineName" placeholder="请输入管道名称">
                         </div>
                     </div>
+                    <div class="form-group">
+                        <label class="col-sm-3 control-label">场内场外</label>
+                        <div class="col-sm-8">
+                            <select class="form-control" id="pipelineArea" name="pipelineArea">
+                                <option value="">请选择</option>
+                                <option value="OUT_SIDE">场外管道</option>
+                                <option value="In_SIDE">场内管道</option>
+                            </select>
+                        </div>
+                    </div>
                     <div class="form-group">
                         <label class="col-sm-3 control-label">直径(mm)</label>
                         <div class="col-sm-8">

src/main/webapp/jsp/pipeline/pipelineDataEdit.jsp

@@ -109,6 +109,16 @@
                                    value="${pipelineData.pipelineName}" placeholder="请输入管道名称">
                         </div>
                     </div>
+                    <div class="form-group">
+                        <label class="col-sm-3 control-label">场内场外</label>
+                        <div class="col-sm-8">
+                            <select class="form-control" id="pipelineArea" name="pipelineArea">
+                                <option value="">请选择</option>
+                                <option value="OUT_SIDE" ${pipelineData.pipelineArea == 'OUT_SIDE' ? 'selected' : ''}>场外管道</option>
+                                <option value="In_SIDE" ${pipelineData.pipelineArea == 'In_SIDE' ? 'selected' : ''}>场内管道</option>
+                            </select>
+                        </div>
+                    </div>
                     <div class="form-group">
                         <label class="col-sm-3 control-label">直径(mm)</label>
                         <div class="col-sm-8">

src/main/webapp/jsp/pipeline/pipelineDataList.jsp

@@ -137,7 +137,8 @@
             page: params.offset / params.limit + 1,
             sort: params.sort,
             order: params.order,
-            search_name: $('#search_name').val()
+            search_name: $('#search_name').val(),
+            search_pipeline_area: $('#search_pipeline_area').val()
         };
     }
 
@@ -169,6 +170,20 @@
                     formatter: function(value, row, index) {
                         return '<span onclick="viewFun(\'' + row.id + '\');" style="color:#004B97;cursor:pointer;">' + row.pipelineName + '</span>';
                     }
+                }, {
+                    field: 'pipelineArea',
+                    title: '场内场外',
+                    align: 'center',
+                    valign: 'middle',
+                    width: '10%',
+                    formatter: function(value, row, index) {
+                        if (value == 'OUT_SIDE') {
+                            return '场外管道';
+                        } else if (value == 'In_SIDE') {
+                            return '场内管道';
+                        }
+                        return value;
+                    }
                 }, {
                     field: 'diameterMm',
                     title: '直径(mm)',
@@ -272,6 +287,11 @@
                                     </button>
                                 </security:authorize>
                                 <div class="form-group pull-right form-inline">
+                                    <select id="search_pipeline_area" name="search_pipeline_area" class="form-control input-sm" style="width: 120px;">
+                                        <option value="">全部区域</option>
+                                        <option value="OUT_SIDE">场外管道</option>
+                                        <option value="In_SIDE">场内管道</option>
+                                    </select>
                                     <div class="input-group input-group-sm" style="width: 250px;">
                                         <input type="text" id="search_name" name="search_name" class="form-control pull-right" placeholder="管道名称">
                                         <div class="input-group-btn">

src/main/webapp/jsp/pipeline/pipelineDataView.jsp

@@ -32,6 +32,18 @@
                             <p class="form-control-static">${pipelineData.pipelineName}</p>
                         </div>
                     </div>
+                    <div class="form-group">
+                        <label class="col-sm-3 control-label">场内场外</label>
+                        <div class="col-sm-8">
+                            <p class="form-control-static">
+                                <c:choose>
+                                    <c:when test="${pipelineData.pipelineArea == 'OUT_SIDE'}">场外管道</c:when>
+                                    <c:when test="${pipelineData.pipelineArea == 'In_SIDE'}">场内管道</c:when>
+                                    <c:otherwise>${pipelineData.pipelineArea}</c:otherwise>
+                                </c:choose>
+                            </p>
+                        </div>
+                    </div>
                     <div class="form-group">
                         <label class="col-sm-3 control-label">直径(mm)</label>
                         <div class="col-sm-8">