2.0 权限以及ES配置

mainFactory/pom.xml

@@ -126,17 +126,17 @@
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>elasticsearch-rest-high-level-client</artifactId>
-            <version>7.17.6</version>
+            <version>7.15.2</version>
         </dependency>
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>elasticsearch-rest-client</artifactId>
-            <version>8.2.1</version>
+            <version>7.15.2</version>
         </dependency>
         <dependency>
             <groupId>org.elasticsearch</groupId>
             <artifactId>elasticsearch</artifactId>
-            <version>8.2.1</version>
+            <version>7.15.2</version>
         </dependency>
 
 

mainFactory/src/main/java/org/bfkj/MainFactoryApplication.java

@@ -290,11 +290,16 @@ public class MainFactoryApplication {
         Runtime.getRuntime().gc();
     }
 
+    private MyDbHelper baseDbHelper;
 
     /*1每2秒执行一次 日志的批量插入*/
     @Scheduled(cron = "*/2 * * * * *")
     public void batchLog() {
         Map<String, List<Object>> logSuccessMap = LogUtils.logSuccessMapList;
+        Map<String, List<Object>> logErrorMap = LogUtils.logErrorMapList;
+        Map<String, List<Object>> logSourceMap = LogUtils.sourceMapList;
+        /*批量写日志*/
+        if (logSuccessMap.size() == 0 && logErrorMap.size() == 0 && logSourceMap.size() == 0) return;
         long current = System.currentTimeMillis();
         List<String> successList = logSuccessMap.keySet().stream().filter(key -> Long.parseLong(key) < current).toList();
         List<Object> logSuccessList = new ArrayList<>();
@@ -302,48 +307,36 @@ public class MainFactoryApplication {
             logSuccessList.addAll(logSuccessMap.get(currentTime));
             logSuccessMap.remove(currentTime);
         }
-        Map<String, List<Object>> logErrorMap = LogUtils.logErrorMapList;
         List<String> errorList = logErrorMap.keySet().stream().filter(key -> Long.parseLong(key) < current).toList();
         List<Object> logErrorList = new ArrayList<>();
         for (String currentTime : errorList) {
             logErrorList.addAll(logErrorMap.get(currentTime));
             logErrorMap.remove(currentTime);
         }
-        Map<String, List<Object>> logSourceMap = LogUtils.sourceMapList;
         List<String> sourceList = logSourceMap.keySet().stream().filter(key -> Long.parseLong(key) < current).toList();
         List<Object> logSourceList = new ArrayList<>();
         for (String currentTime : sourceList) {
             logSourceList.addAll(logSourceMap.get(currentTime));
             logSourceMap.remove(currentTime);
         }
-        /*批量写日志*/
-        if (logErrorList.size() == 0 && logSourceList.size() == 0 && logSuccessList.size() == 0) return;
-        MyDbHelper myDbHelper = getDbHelper();
-        if (Objects.isNull(myDbHelper)){
+        if (Objects.isNull(baseDbHelper)) {
+            baseDbHelper = new MyDbHelper(AppConfig.getSystemParams(AppConfig.REMOTE_DB_CONNECT));//获取底座数据库对象
+            if (Objects.nonNull(baseDbHelper.getErrorMessage())) {
+                return;
+            }
+        }
+        if (Objects.isNull(baseDbHelper)) {
             System.out.println("获取远程数据库失败：写入日志失败");
             return;
         }
         try {
-            if (logSuccessList.size() > 0) myDbHelper.JDBCBatch(INSERT_SQL, logSuccessList, null);
-            if (logErrorList.size() > 0) myDbHelper.JDBCBatch(ERROR_SQL, logErrorList, null);
+            if (logSuccessList.size() > 0) baseDbHelper.JDBCBatch(INSERT_SQL, logSuccessList, null);
+            if (logErrorList.size() > 0) baseDbHelper.JDBCBatch(ERROR_SQL, logErrorList, null);
             if (logSourceList.size() > 0) {
-                myDbHelper.JDBCBatch(FULL_DATA_SQL, logSourceList, null);
+                baseDbHelper.JDBCBatch(FULL_DATA_SQL, logSourceList, null);
             }
         } catch (Exception e) {
             System.out.println("日志记录异常" + LogUtils.getException(e));
         }
     }
-
-    private MyDbHelper baseDbHelper;
-
-    public MyDbHelper getDbHelper() {
-        if (Objects.isNull(baseDbHelper)) {
-            baseDbHelper = new MyDbHelper(AppConfig.getSystemParams(AppConfig.REMOTE_DB_CONNECT));//获取底座数据库对象
-            if (Objects.nonNull(baseDbHelper.getErrorMessage())) {
-                return null;
-            }
-        }
-        return baseDbHelper;
-
-    }
 }

mainFactory/src/main/java/org/bfkj/application/DataProcess.java

@@ -186,11 +186,11 @@ public class DataProcess {//数据处理对象
                             signRowAuth.put(columnName, row_auth);
                             rowAuth.addAll(baseDbHelper.changeSignFilter(signRowAuth, connect, connect.equals("!=") ? "and" : "or"));
                         }
-                        if (rowAuth.size() > 0) {//存在行权限则进行括号包裹
-                            rowAuth.get(0).put("left", "(");
-                            rowAuth.get(rowAuth.size() - 1).put("connector", "  and ");
-                            rowAuth.get(rowAuth.size() - 1).put("right", " ) ");
-                        }   //(c1 = pek )
+//                        if (rowAuth.size() > 0) {//存在行权限则进行括号包裹
+//                            rowAuth.get(0).put("left", "(");
+//                            rowAuth.get(rowAuth.size() - 1).put("connector", "  and ");
+//                            rowAuth.get(rowAuth.size() - 1).put("right", " ) ");
+//                        }   //(c1 = pek )
                     }
                 });
                 if (rowAuth.size() > 0) {//存在行权限则进行括号包裹

mainFactory/src/main/java/org/bfkj/config/EsConfig.java

@@ -0,0 +1,116 @@
+package org.bfkj.config;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class EsConfig {
+    @Value("${isEs.open}")
+    private boolean open;
+    @Value("${isEs.scheme}")
+    private String scheme;
+    @Value("${isEs.hostname}")
+    private String hostname;
+    @Value("${isEs.port}")
+    private Integer port;
+    @Value("${isEs.auth}")
+    private Boolean auth;
+    @Value("${isEs.username}")
+    private String username;
+    @Value("${isEs.password}")
+    private String password;
+    @Value("${isEs.successIndexName}")
+    private String successIndexName;
+
+    @Value("${isEs.errorIndexName}")
+    private String errorIndexName;
+
+    @Bean("EsConfig")
+    public EsConfig EsConfig() {
+        EsConfig esConfig = new EsConfig();
+        esConfig.open = open;
+        esConfig.scheme = scheme;
+        esConfig.hostname = hostname;
+        esConfig.port = port;
+        esConfig.auth = auth;
+        esConfig.username = username;
+        esConfig.password = password;
+        esConfig.errorIndexName = errorIndexName;
+        esConfig.successIndexName = successIndexName;
+        return esConfig;
+
+    }
+
+    public boolean isOpen() {
+        return open;
+    }
+
+    public void setOpen(boolean open) {
+        this.open = open;
+    }
+
+    public String getScheme() {
+        return scheme;
+    }
+
+    public void setScheme(String scheme) {
+        this.scheme = scheme;
+    }
+
+    public String getHostname() {
+        return hostname;
+    }
+
+    public void setHostname(String hostname) {
+        this.hostname = hostname;
+    }
+
+    public Integer getPort() {
+        return port;
+    }
+
+    public void setPort(Integer port) {
+        this.port = port;
+    }
+
+    public Boolean getAuth() {
+        return auth;
+    }
+
+    public void setAuth(Boolean auth) {
+        this.auth = auth;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public String getSuccessIndexName() {
+        return successIndexName;
+    }
+
+    public void setSuccessIndexName(String successIndexName) {
+        this.successIndexName = successIndexName;
+    }
+
+    public String getErrorIndexName() {
+        return errorIndexName;
+    }
+
+    public void setErrorIndexName(String errorIndexName) {
+        this.errorIndexName = errorIndexName;
+    }
+}

mainFactory/src/main/java/org/bfkj/protocol/ElasticHandler.java

@@ -5,18 +5,18 @@ import org.apache.http.auth.AuthScope;
 import org.apache.http.auth.UsernamePasswordCredentials;
 import org.apache.http.client.CredentialsProvider;
 import org.apache.http.impl.client.BasicCredentialsProvider;
-import org.bfkj.config.AppConfig;
 import org.bfkj.utils.LogUtils;
 import org.bfkj.utils.MapTools;
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.common.xcontent.XContentType;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -28,18 +28,21 @@ import java.util.stream.Collectors;
 /*es协议*/
 public class ElasticHandler {
 
-    private RestHighLevelClient restHighLevelClient;
+    private static RestHighLevelClient restHighLevelClient;
 
-    public synchronized Map<String, Object> sendData(String indexName, List<Object> dataContent, String connectConfig, String serviceId) {
+    public boolean sendData(String indexName, List<Object> dataContent, String connectConfig, String serviceId) {
         if (Objects.isNull(indexName)) {
-            return MapTools.processFail("队列为空: " + dataContent);
+            MapTools.processFail("队列为空: " + dataContent);
+            return true;
         }
         if (Objects.isNull(dataContent)) {
-            return MapTools.processSuccess(null);
+            MapTools.processSuccess(null);
+            return true;
         }
-        try {
-            Map<String, Object> connectConfigMaps = (Map<String, Object>) MapTools.strToObj(connectConfig);
-            if (Objects.isNull(restHighLevelClient)) {
+
+        if (Objects.isNull(restHighLevelClient) ){
+            try {
+                Map<String, Object> connectConfigMaps = (Map<String, Object>) MapTools.strToObj(connectConfig);
                 String scheme = connectConfigMaps.get("scheme").toString();
                 String hostname = connectConfigMaps.get("hostname").toString();
                 int port = Integer.parseInt(connectConfigMaps.get("port").toString());
@@ -56,40 +59,45 @@ public class ElasticHandler {
                     //无权限验证
                     restHighLevelClient = new RestHighLevelClient(RestClient.builder(new HttpHost(hostname, port, scheme)));
                 }
+            } catch (Exception e) {
+                MapTools.processFail("es发送参数异常：" + LogUtils.getException(e));
+                return false;
             }
-        } catch (Exception e) {
-            return MapTools.processFail("es发送参数异常：" + LogUtils.getException(e));
         }
         String dataJSON = "";
         try {
-            IndexRequest request = new IndexRequest(indexName);
-            dataJSON = MapTools.objToJSONStr(dataContent);
-            request.id(createLifeCycleCol(Long.valueOf(AppConfig.WORK_ID), Integer.valueOf(MapTools.isBlank(serviceId) ? "0" : serviceId)));
-            request.timeout("1s");
-            request.source(dataJSON, XContentType.JSON);
-            IndexResponse indexResponse = restHighLevelClient.index(request, RequestOptions.DEFAULT);
-            ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
-            if (shardInfo.getFailed() > 0) {
-                String collect = Arrays.stream(shardInfo.getFailures()).map(ShardOperationFailedException::reason).collect(Collectors.joining(";\n"));
-                return MapTools.processSuccess(collect);
+            if (dataContent.size() > 0) {
+                IndexRequest request = new IndexRequest(indexName);
+                for (Object data : dataContent) {
+                    dataJSON = MapTools.objToJSONStr(data);
+                    request.id(createLifeCycleCol(1L, Integer.valueOf(MapTools.isBlank(serviceId) ? "0" : serviceId)));
+                    request.timeout("1s");
+                    request.source(dataJSON, XContentType.JSON);
+                    request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+                    IndexResponse indexResponse = restHighLevelClient.index(request, RequestOptions.DEFAULT);
+                    ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
+                    if (shardInfo.getFailed() > 0) {
+                        String collect = Arrays.stream(shardInfo.getFailures()).map(ShardOperationFailedException::reason).collect(Collectors.joining(";\n"));
+                        MapTools.processSuccess(collect);
+                        return false;
+                    }
+                }
             }
         } catch (IOException e) {
             try {
-                Thread.sleep(1000 * 6);
-            } catch (InterruptedException ex) {
+                restHighLevelClient.close();
+            } catch (IOException ex) {
                 System.out.println("es数据发送异常： ".concat(LogUtils.getException(e)));
             }
-            return MapTools.processFail(LogUtils.getException(e));
-        } finally {
             try {
-                if (null != restHighLevelClient) {
-                    restHighLevelClient.close();
-                }
-            } catch (IOException ex) {
-                System.out.println("es数据发送异常： ".concat(LogUtils.getException(ex)));
+                Thread.sleep(1000 * 6);
+            } catch (InterruptedException ex) {
+                System.out.println("es数据发送异常： ".concat(LogUtils.getException(e)));
             }
+            restHighLevelClient = null;
+            return  false;
         }
-        return MapTools.processSuccess("数据发送成功：".concat(dataJSON));
+        return true;
     }
 
     private long sequence = 0L;

mainFactory/src/main/java/org/bfkj/utils/LogUtils.java

@@ -2,7 +2,9 @@ package org.bfkj.utils;
 
 
 import org.bfkj.config.AppConfig;
-import org.bfkj.config.ObjectMap;
+import org.bfkj.config.EsConfig;
+import org.bfkj.config.SpringContextBeans;
+import org.bfkj.protocol.ElasticHandler;
 
 import java.text.SimpleDateFormat;
 import java.util.*;
@@ -16,6 +18,7 @@ public class LogUtils { // 依据服务编号ID获取服务类型，如果是异
     public static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
     private static Long deleteTime = 0L;
 
+    private static MyDbHelper myDbHelper;
 
     public static void log(String location, String success, Object calculationLocation, String logContent, String serviceId, Object iNDataContent, String dataObjectId, Object OutDataContent, Object event) {
         String inData = MapTools.objToJSONStr(iNDataContent);
@@ -28,6 +31,36 @@ public class LogUtils { // 依据服务编号ID获取服务类型，如果是异
             System.out.println("过滤：读取空消息： 出现错误".concat(LogUtils.getException(e)).concat(" iNDataContent: ").concat(inData == null ? "" : inData));
             return;
         }
+        EsConfig esConfig = SpringContextBeans.getBean("EsConfig");
+        if (esConfig.isOpen()) {
+            HashMap<Object, Object> esSendData = new HashMap<>();
+            esSendData.put("success", success);
+            esSendData.put("location", location);
+            esSendData.put("logContent", logContent);
+            esSendData.put("ctime", new Date());
+            esSendData.put("serviceId", serviceId);
+            esSendData.put("workId", AppConfig.WORK_ID);
+            esSendData.put("event", Objects.isNull(event) ? null : event.toString());
+            esSendData.put("iNDataContent", inData);
+            esSendData.put("outDataContent", MapTools.objToJSONStr(OutDataContent));
+            esSendData.put("calculationLocation", Objects.isNull(calculationLocation) ? null : calculationLocation.toString());
+            esSendData.put("dataObjectId", dataObjectId);
+            ElasticHandler elasticHandler = new ElasticHandler();
+            String connectConfig = MapTools.objToJSONStr(esConfig);
+            boolean b;
+            if (success.equals("0")) {
+                b = elasticHandler.sendData(esConfig.getSuccessIndexName(), new ArrayList<>() {{
+                    add(esSendData);
+                }}, connectConfig, serviceId);
+            } else {
+                b = elasticHandler.sendData(esConfig.getErrorIndexName(), new ArrayList<>() {{
+                    add(esSendData);
+                }}, connectConfig, serviceId);
+            }
+            if (b) {
+                return;
+            }
+        }
         String shortData = MapTools.isNotBlank(inData) ? (inData.length() > 20480 ? inData.substring(0, 20480) : inData) : null;
         List<Object> logParam = new ArrayList<>() {{
             add(success);
@@ -64,16 +97,18 @@ public class LogUtils { // 依据服务编号ID获取服务类型，如果是异
             sourceMapList.put(key, sourceLogList);
         }
         if (MapTools.isNotBlank(serviceId) && (System.currentTimeMillis() - deleteTime) > 10000) {
-            MyDbHelper myDbHelper = ObjectMap.getordropMyDbHelper(AppConfig.getSystemParams(AppConfig.REMOTE_DB_CONNECT));
+            if (null == myDbHelper) {
+                myDbHelper = new MyDbHelper(AppConfig.getSystemParams(AppConfig.REMOTE_DB_CONNECT));//获取底座数据库对象;
+            }
             if (Objects.nonNull(myDbHelper.getErrorMessage())) {
                 System.out.println("记录日志： 获取 MyDbHelper对象失败,日志信息 ：" + myDbHelper.getErrorMessage());
+                return;
             }
             myDbHelper.updateByCondition("delete log_success_a where serviceId =? and id not in (select id  from log_success_a  where serviceId = ? order by id desc limit 100 ) ", null, serviceId, serviceId);
             deleteTime = System.currentTimeMillis();
         }
     }
 
-
     /**
      * 获取当前行
      *

mainFactory/src/main/java/org/bfkj/utils/MyDbHelper.java

@@ -40,7 +40,7 @@ public class MyDbHelper {//目前只差主键的条件优化
         put("3", "delete from 表名 ");
     }};
     //还差一个基本字段列表，避免动态表自动生成(event,page,pagesize,serviceid)
-    private static String errorMessage = null; //当前数据输入处理对象不可用信息
+    private  String errorMessage = null; //当前数据输入处理对象不可用信息
 
     public MyDbHelper(String connectStr) { // 需要json字符串
         Map<String, Object> connectMaps = null;
@@ -189,7 +189,7 @@ public class MyDbHelper {//目前只差主键的条件优化
                         }
                     }
                     sqlParmMap.put("dbParam", dbParam);//执行参数
-                    if (MapTools.isBlank(whereStr) && "2,3".contains(newEvent)) {
+                    if (dbParam.size() ==0 &&MapTools.isBlank(whereStr) && "2,3".contains(newEvent)) {
                         sqlParmMap.put("noExec", " 禁止全表更新、删除");//执行参数
                     }
                     execSQLMap.put(execSQLStr, sqlParmMap);//添加到最终执行列表中
@@ -495,7 +495,7 @@ public class MyDbHelper {//目前只差主键的条件优化
                         if (Objects.nonNull(column) && !varList.contains(column.toString())) {//列名不为空 且 (可用列未获取 或者 当前列存在于可用列)
                             String connector = Objects.isNull(filterMap.get("connector")) ? "" : filterMap.get("connector").toString();//处理连接字符串
                             Object comparator = Objects.isNull(filterMap.get("comparator")) ? "" : filterMap.get("comparator").toString();//比较符
-                            whereStr = whereStr.concat(filterMap.get("left").toString()).concat(column.toString()).concat(comparator.toString()).concat(" is null ".equals(comparator) ? "" : "?").concat(filterMap.get("right").toString()).concat((MapTools.isBlank(connector)) ? " and " : ("lastConnector".equals(connector) ? "" : " ".concat(connector).concat(" ")));
+                            whereStr = whereStr.concat(filterMap.get("left").toString()).concat(column.toString()).concat(comparator.toString()).concat(" is null ".equals(comparator) ? "" : "?").concat(filterMap.get("right").toString()).concat((MapTools.isBlank(connector)) ? " and " : ("lastConnector".equals(connector) ? " " : " ".concat(connector).concat(" ")));
                             if (!" is null ".equals(comparator)) {//不是is null，且不是SQL语句执行方式
                                 dbFilter.add(filterMap.get("value"));//添加执行参数
                             }

mainFactory/src/main/resources/ES索引创建命令

@@ -0,0 +1,51 @@
+{
+    "settings": {
+        "refresh_interval": "-1"
+    },
+    "mappings": {
+        "properties": {
+            "ctime": {
+                "type": "date"
+            },
+            "location": {
+                "type": "text"
+            },
+            "dataObjectId": {
+                "type": "text",
+                "fields": {
+                    "keyword": {
+                        "type": "keyword",
+                        "ignore_above": 256
+                    }
+                }
+            },
+            "logContent": {
+                "type": "text"
+            },
+             "success": {
+                "type": "text"
+            },
+             "serviceId": {
+                "type": "text"
+            },
+            "workId": {
+                "type": "text"
+            },
+            "event": {
+                "type": "text"
+            },
+             "outDataContent": {
+                "type": "text"
+            },
+            "iNDataContent": {
+                "type": "text"
+            },
+            "calculationLocation": {
+                "type": "text"
+            },
+            "dataobjectid": {
+                "type": "keyword"
+            }
+        }
+    }
+}

mainFactory/src/main/resources/application.yml

@@ -22,6 +22,36 @@ mydb:
   type: ${MYDB_DRIVER:com.mysql.cj.jdbc.Driver}
   serviceURL: ${MY_SERVICE_URL:127.0.0.1:8080}
 
+#mydb:
+#  url: ${MYDB_URL:jdbc:mysql://rm-bp12c6mmk845m73p48o.mysql.rds.aliyuncs.com:3306/test_cen}
+#  title: ${MYDB_USER:bf_dev_epi}
+#  cipher: ${MYDB_PASSWD:Bfepi2021(}
+#  type: ${MYDB_DRIVER:com.mysql.cj.jdbc.Driver}
+#  serviceURL: ${MY_SERVICE_URL:172.23.61.4:9999}
+
+#isEs:
+#  open: true
+#  scheme: http
+#  hostname: 120.26.64.82
+#  port: 9200
+#  auth: false
+#  username: ""
+#  password: ""
+#  successIndexName: "success_log_center2"
+#  errorIndexName: "error_log_center2"
+
+isEs:
+  open: false
+  scheme: http
+  hostname: 10.211.67.175
+  port: 9201
+  auth: true
+  username: "elastic"
+  password: "TUihk5^a0J+GkI4!4"
+  successIndexName: "success_log_center"
+  errorIndexName: "error_log_center"
+
+
 service:
   pool:
     maxPoolSize: 512