From 5f6650b5c81eaaa95e7f273fb00a84c0200e9f75 Mon Sep 17 00:00:00 2001 From: houxinyu Date: Wed, 20 Sep 2023 20:40:31 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96SQL=E8=BD=AE=E8=AF=A2&?= =?UTF-8?q?=E6=8A=BD=E5=8F=96=E7=BB=9F=E4=B8=80=E6=AD=A3=E5=88=99=E8=BD=AC?= =?UTF-8?q?=E6=8D=A2=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../parser/helper/NodeConvertHelper.java | 108 +++++++++++ .../parser/apollo/util/ApolloParseHelper.java | 1 + .../parser/redis/mode/RedisParserHelper.java | 72 +------- .../mode/polling/RedisParserPollingMode.java | 3 +- .../redis/mode/polling/ScriptPollingTask.java | 3 +- .../subscribe/RedisParserSubscribeMode.java | 5 +- .../parser/sql/util/ChainPollingTask.java | 78 +++----- .../liteflow/parser/sql/util/JDBCHelper.java | 4 +- .../parser/sql/util/ScriptPollingTask.java | 168 +++--------------- .../parser/zk/util/ZkParserHelper.java | 81 +-------- 10 files changed, 172 insertions(+), 351 deletions(-) create mode 100644 liteflow-core/src/main/java/com/yomahub/liteflow/parser/helper/NodeConvertHelper.java diff --git a/liteflow-core/src/main/java/com/yomahub/liteflow/parser/helper/NodeConvertHelper.java b/liteflow-core/src/main/java/com/yomahub/liteflow/parser/helper/NodeConvertHelper.java new file mode 100644 index 000000000..b4309eaa9 --- /dev/null +++ b/liteflow-core/src/main/java/com/yomahub/liteflow/parser/helper/NodeConvertHelper.java @@ -0,0 +1,108 @@ +package com.yomahub.liteflow.parser.helper; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.util.ReUtil; +import cn.hutool.core.util.StrUtil; +import com.yomahub.liteflow.builder.LiteFlowNodeBuilder; +import com.yomahub.liteflow.enums.NodeTypeEnum; + +import java.util.List; + +/** + * 冒号形式节点NodeSimpleVO转换 通用 Helper + * + * @author hxinyu + */ +public class NodeConvertHelper { + + /*script节点的修改/添加*/ + public static void changeScriptNode(NodeSimpleVO nodeSimpleVO, String newValue) { + // 有语言类型 + if (StrUtil.isNotBlank(nodeSimpleVO.getLanguage())) { + LiteFlowNodeBuilder.createScriptNode() + .setId(nodeSimpleVO.getNodeId()) + .setType(NodeTypeEnum.getEnumByCode(nodeSimpleVO.getType())) + .setName(nodeSimpleVO.getName()) + .setScript(newValue) + .setLanguage(nodeSimpleVO.getLanguage()) + .build(); + } + // 没有语言类型 + else { + LiteFlowNodeBuilder.createScriptNode() + .setId(nodeSimpleVO.getNodeId()) + .setType(NodeTypeEnum.getEnumByCode(nodeSimpleVO.getType())) + .setName(nodeSimpleVO.getName()) + .setScript(newValue) + .build(); + } + } + + public static NodeSimpleVO convert(String scriptKey){ + // 不需要去理解这串正则，就是一个匹配冒号的 + // 一定得是a:b，或是a:b:c...这种完整类型的字符串的 + List matchItemList = ReUtil.findAllGroup0("(?<=[^:]:)[^:]+|[^:]+(?=:[^:])", scriptKey); + if (CollUtil.isEmpty(matchItemList)) { + return null; + } + NodeSimpleVO nodeSimpleVO = new NodeSimpleVO(); + if (matchItemList.size() > 1) { + nodeSimpleVO.setNodeId(matchItemList.get(0)); + nodeSimpleVO.setType(matchItemList.get(1)); + } + + if (matchItemList.size() > 2) { + nodeSimpleVO.setName(matchItemList.get(2)); + } + + if (matchItemList.size() > 3) { + nodeSimpleVO.setLanguage(matchItemList.get(3)); + } + + return nodeSimpleVO; + } + + + public static class NodeSimpleVO { + + private String nodeId; + + private String type; + + private String name = StrUtil.EMPTY; + + private String language; + + public String getNodeId() { + return nodeId; + } + + public void setNodeId(String nodeId) { + this.nodeId = nodeId; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getLanguage() { + return language; + } + + public void setLanguage(String language) { + this.language = language; + } + } +} \ No newline at end of file diff --git a/liteflow-rule-plugin/liteflow-rule-apollo/src/main/java/com/yomahub/liteflow/parser/apollo/util/ApolloParseHelper.java b/liteflow-rule-plugin/liteflow-rule-apollo/src/main/java/com/yomahub/liteflow/parser/apollo/util/ApolloParseHelper.java index 468704be1..b8a4511c4 100644 --- a/liteflow-rule-plugin/liteflow-rule-apollo/src/main/java/com/yomahub/liteflow/parser/apollo/util/ApolloParseHelper.java +++ b/liteflow-rule-plugin/liteflow-rule-apollo/src/main/java/com/yomahub/liteflow/parser/apollo/util/ApolloParseHelper.java @@ -15,6 +15,7 @@ import com.yomahub.liteflow.enums.NodeTypeEnum; import com.yomahub.liteflow.flow.FlowBus; import com.yomahub.liteflow.parser.apollo.exception.ApolloException; import com.yomahub.liteflow.parser.apollo.vo.ApolloParserConfigVO; +import com.yomahub.liteflow.parser.helper.NodeConvertHelper; import com.yomahub.liteflow.spi.holder.ContextAwareHolder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/RedisParserHelper.java b/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/RedisParserHelper.java index eb60d2cdd..345fc9e2e 100644 --- a/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/RedisParserHelper.java +++ b/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/RedisParserHelper.java @@ -8,6 +8,7 @@ import com.yomahub.liteflow.builder.LiteFlowNodeBuilder; import com.yomahub.liteflow.enums.NodeTypeEnum; import com.yomahub.liteflow.log.LFLog; import com.yomahub.liteflow.log.LFLoggerManager; +import com.yomahub.liteflow.parser.helper.NodeConvertHelper; import com.yomahub.liteflow.parser.redis.vo.RedisParserVO; import org.redisson.config.Config; import org.redisson.config.SentinelServersConfig; @@ -112,7 +113,7 @@ public interface RedisParserHelper { * @param newValue 新的script值 */ static void changeScriptNode(String scriptFieldValue, String newValue) { - NodeSimpleVO nodeSimpleVO = convert(scriptFieldValue); + NodeConvertHelper.NodeSimpleVO nodeSimpleVO = NodeConvertHelper.convert(scriptFieldValue); // 有语言类型 if (StrUtil.isNotBlank(nodeSimpleVO.getLanguage())) { LiteFlowNodeBuilder.createScriptNode() @@ -133,73 +134,4 @@ public interface RedisParserHelper { .build(); } } - - static NodeSimpleVO convert(String str) { - // 不需要去理解这串正则，就是一个匹配冒号的 - // 一定得是a:b，或是a:b:c...这种完整类型的字符串的 - List matchItemList = ReUtil.findAllGroup0("(?<=[^:]:)[^:]+|[^:]+(?=:[^:])", str); - if (CollUtil.isEmpty(matchItemList)) { - return null; - } - - NodeSimpleVO nodeSimpleVO = new NodeSimpleVO(); - if (matchItemList.size() > 1) { - nodeSimpleVO.setNodeId(matchItemList.get(0)); - nodeSimpleVO.setType(matchItemList.get(1)); - } - - if (matchItemList.size() > 2) { - nodeSimpleVO.setName(matchItemList.get(2)); - } - - if (matchItemList.size() > 3) { - nodeSimpleVO.setLanguage(matchItemList.get(3)); - } - - return nodeSimpleVO; - } - - class NodeSimpleVO { - - private String nodeId; - - private String type; - - private String name = StrUtil.EMPTY; - - private String language; - - public String getNodeId() { - return nodeId; - } - - public void setNodeId(String nodeId) { - this.nodeId = nodeId; - } - - public String getType() { - return type; - } - - public void setType(String type) { - this.type = type; - } - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - public String getLanguage() { - return language; - } - - public void setLanguage(String language) { - this.language = language; - } - - } } diff --git a/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/polling/RedisParserPollingMode.java b/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/polling/RedisParserPollingMode.java index 662b801b5..352bf1f9e 100644 --- a/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/polling/RedisParserPollingMode.java +++ b/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/polling/RedisParserPollingMode.java @@ -6,6 +6,7 @@ import cn.hutool.core.thread.NamedThreadFactory; import cn.hutool.core.util.ObjectUtil; import cn.hutool.core.util.StrUtil; import cn.hutool.crypto.digest.DigestUtil; +import com.yomahub.liteflow.parser.helper.NodeConvertHelper; import com.yomahub.liteflow.parser.redis.exception.RedisException; import com.yomahub.liteflow.parser.redis.mode.RClient; import com.yomahub.liteflow.parser.redis.mode.RedisMode; @@ -148,7 +149,7 @@ public class RedisParserPollingMode implements RedisParserHelper { List scriptItemContentList = new ArrayList<>(); for (String scriptFieldValue : scriptFieldSet) { - NodeSimpleVO nodeSimpleVO = RedisParserHelper.convert(scriptFieldValue); + NodeConvertHelper.NodeSimpleVO nodeSimpleVO = NodeConvertHelper.convert(scriptFieldValue); if (ObjectUtil.isNull(nodeSimpleVO)) { throw new RedisException( StrUtil.format("The name of the redis field [{}] in scriptKey [{}] is invalid", diff --git a/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/polling/ScriptPollingTask.java b/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/polling/ScriptPollingTask.java index a23965c37..40d4704e6 100644 --- a/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/polling/ScriptPollingTask.java +++ b/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/polling/ScriptPollingTask.java @@ -5,6 +5,7 @@ import cn.hutool.crypto.digest.DigestUtil; import com.yomahub.liteflow.flow.FlowBus; import com.yomahub.liteflow.log.LFLog; import com.yomahub.liteflow.log.LFLoggerManager; +import com.yomahub.liteflow.parser.helper.NodeConvertHelper; import com.yomahub.liteflow.parser.redis.mode.RClient; import com.yomahub.liteflow.parser.redis.mode.RedisParserHelper; import com.yomahub.liteflow.parser.redis.vo.RedisParserVO; @@ -68,7 +69,7 @@ public class ScriptPollingTask implements Runnable { String newSHA = scriptClient.evalSha(valueLua, scriptKey, scriptFieldValue); if (StrUtil.equals(newSHA, "nil")) { //新SHA值为nil, 即未获取到该script,表示该script已被删除 - RedisParserHelper.NodeSimpleVO nodeSimpleVO = RedisParserHelper.convert(scriptFieldValue); + NodeConvertHelper.NodeSimpleVO nodeSimpleVO = NodeConvertHelper.convert(scriptFieldValue); FlowBus.getNodeMap().remove(nodeSimpleVO.getNodeId()); LOG.info("starting reload flow config... delete key={}", scriptFieldValue); diff --git a/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/subscribe/RedisParserSubscribeMode.java b/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/subscribe/RedisParserSubscribeMode.java index b115baec5..aff895cd9 100644 --- a/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/subscribe/RedisParserSubscribeMode.java +++ b/liteflow-rule-plugin/liteflow-rule-redis/src/main/java/com/yomahub/liteflow/parser/redis/mode/subscribe/RedisParserSubscribeMode.java @@ -6,6 +6,7 @@ import cn.hutool.core.util.ObjectUtil; import cn.hutool.core.util.StrUtil; import com.yomahub.liteflow.builder.el.LiteFlowChainELBuilder; import com.yomahub.liteflow.flow.FlowBus; +import com.yomahub.liteflow.parser.helper.NodeConvertHelper; import com.yomahub.liteflow.parser.redis.exception.RedisException; import com.yomahub.liteflow.parser.redis.mode.RClient; import com.yomahub.liteflow.parser.redis.mode.RedisMode; @@ -109,7 +110,7 @@ public class RedisParserSubscribeMode implements RedisParserHelper { for (Map.Entry entry : scriptMap.entrySet()) { String scriptFieldValue = entry.getKey(); String scriptData = entry.getValue(); - NodeSimpleVO nodeSimpleVO = RedisParserHelper.convert(scriptFieldValue); + NodeConvertHelper.NodeSimpleVO nodeSimpleVO = NodeConvertHelper.convert(scriptFieldValue); if (ObjectUtil.isNull(nodeSimpleVO)) { throw new RedisException( StrUtil.format("The name of the redis field [{}] in scriptKey [{}] is invalid", @@ -193,7 +194,7 @@ public class RedisParserSubscribeMode implements RedisParserHelper { //删除 script scriptClient.addListener(scriptKey, (EntryRemovedListener) event -> { LOG.info("starting reload flow config... delete key={}", event.getKey()); - NodeSimpleVO nodeSimpleVO = RedisParserHelper.convert(event.getKey()); + NodeConvertHelper.NodeSimpleVO nodeSimpleVO = NodeConvertHelper.convert(event.getKey()); FlowBus.getNodeMap().remove(nodeSimpleVO.getNodeId()); }); } diff --git a/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/ChainPollingTask.java b/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/ChainPollingTask.java index cac3bf033..a8de50f4f 100644 --- a/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/ChainPollingTask.java +++ b/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/ChainPollingTask.java @@ -1,6 +1,7 @@ package com.yomahub.liteflow.parser.sql.util; import cn.hutool.core.util.StrUtil; +import cn.hutool.crypto.digest.DigestUtil; import com.yomahub.liteflow.builder.el.LiteFlowChainELBuilder; import com.yomahub.liteflow.flow.FlowBus; import com.yomahub.liteflow.log.LFLog; @@ -29,12 +30,6 @@ public class ChainPollingTask implements Runnable { private static final String NEW_CHAIN_PATTERN = "SELECT {} FROM {} WHERE {}=? AND {}=?"; - private static final String SHA_PATTERN = "SHA1({}) AS SHA1"; - - private static final String SHA_PATTERN_FOR_H2 = "RAWTOHEX(HASH('SHA-1', {})) AS SHA1"; - - private static final String SHA_FIELD_NAME = "SHA1"; - public static Connection conn; private SQLParserVO sqlParserVO; @@ -53,6 +48,8 @@ public class ChainPollingTask implements Runnable { @Override public void run() { + PreparedStatement stmt = null; + ResultSet rs = null; try{ String chainTableName = sqlParserVO.getChainTableName(); String elDataField = sqlParserVO.getElDataField(); @@ -60,52 +57,37 @@ public class ChainPollingTask implements Runnable { String chainApplicationNameField = sqlParserVO.getChainApplicationNameField(); String applicationName = sqlParserVO.getApplicationName(); - String SHAField = StrUtil.format(SHA_PATTERN, elDataField); - //h2数据库计算SHA的函数与MySQL不同 - if(StrUtil.equals(sqlParserVO.getDriverClassName(), "org.h2.Driver")){ - SHAField = StrUtil.format(SHA_PATTERN_FOR_H2, elDataField); - } - - String sqlCmd = StrUtil.format(SQL_PATTERN, chainNameField, SHAField, chainTableName, + String sqlCmd = StrUtil.format(SQL_PATTERN, chainNameField, elDataField, chainTableName, chainApplicationNameField); - PreparedStatement stmt = conn.prepareStatement(sqlCmd, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt = conn.prepareStatement(sqlCmd, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); // 设置游标拉取数量 stmt.setFetchSize(FETCH_SIZE_MAX); stmt.setString(1, applicationName); - ResultSet rs = stmt.executeQuery(); + rs = stmt.executeQuery(); Set newChainSet = new HashSet<>(); while(rs.next()) { String chainName = getStringFromResultSet(rs, chainNameField); - String newSHA = getStringFromResultSet(rs, SHA_FIELD_NAME); + String newData = getStringFromResultSet(rs, elDataField); + String newSHA = DigestUtil.sha1Hex(newData); newChainSet.add(chainName); //如果封装的SHAMap中不存在该chain, 表示该chain为新增 if(!chainSHAMap.containsKey(chainName)) { - //获取新chain结果 - ResultSet newChainRS = getNewChainRS(elDataField, chainTableName, chainNameField, - chainApplicationNameField, applicationName, chainName); - if(newChainRS.next()) { - String newELData = getStringFromResultSet(newChainRS, elDataField); - //新增chain - LiteFlowChainELBuilder.createChain().setChainId(chainName).setEL(newELData).build(); - LOG.info("starting reload flow config... create chain={}, new value={},", chainName, newELData); - //加入到shaMap - chainSHAMap.put(chainName, newSHA); - } + //新增chain + LiteFlowChainELBuilder.createChain().setChainId(chainName).setEL(newData).build(); + LOG.info("starting reload flow config... create chain={}, new value={},", chainName, newData); + //加入到shaMap + chainSHAMap.put(chainName, newSHA); + } else if (!StrUtil.equals(newSHA, chainSHAMap.get(chainName))) { //SHA值发生变化,表示该chain的值已被修改,重新拉取变化的chain - ResultSet newChainRS = getNewChainRS(elDataField, chainTableName, chainNameField, - chainApplicationNameField, applicationName, chainName); - if(newChainRS.next()) { - String newELData = getStringFromResultSet(newChainRS, elDataField); - //修改chain - LiteFlowChainELBuilder.createChain().setChainId(chainName).setEL(newELData).build(); - LOG.info("starting reload flow config... update chain={}, new value={},", chainName, newELData); - //修改shaMap - chainSHAMap.put(chainName, newSHA); - } + //修改chain + LiteFlowChainELBuilder.createChain().setChainId(chainName).setEL(newData).build(); + LOG.info("starting reload flow config... update chain={}, new value={},", chainName, newData); + //修改shaMap + chainSHAMap.put(chainName, newSHA); } //SHA值无变化,表示该chain未改变 } @@ -128,28 +110,14 @@ public class ChainPollingTask implements Runnable { } } } - } - catch (Exception e) { + } catch (Exception e) { LOG.error("[Exception during SQL chain polling] " + e.getMessage(), e); + } finally { + // 关闭连接 + LiteFlowJdbcUtil.close(null, stmt, rs); } } - private ResultSet getNewChainRS(String elDataField, String chainTableName, String chainNameField, - String chainApplicationNameField, String applicationName, String chainName) { - ResultSet rs = null; - String sqlCmd = StrUtil.format(NEW_CHAIN_PATTERN, elDataField, chainTableName, - chainNameField, chainApplicationNameField); - try{ - PreparedStatement stmt = conn.prepareStatement(sqlCmd, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); - stmt.setString(1, chainName); - stmt.setString(2, applicationName); - rs = stmt.executeQuery(); - }catch (Exception e) { - throw new ELSQLException(e.getMessage()); - } - return rs; - } - private String getStringFromResultSet(ResultSet rs, String field) throws SQLException { String data = rs.getString(field); if (StrUtil.isBlank(data)) { diff --git a/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/JDBCHelper.java b/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/JDBCHelper.java index 5b9c2083b..5be349ed7 100644 --- a/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/JDBCHelper.java +++ b/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/JDBCHelper.java @@ -60,10 +60,10 @@ public class JDBCHelper { private static ScheduledThreadPoolExecutor pollExecutor; //chain的SHA1加密值 用于轮询时确定chain是否变化 - private Map chainSHAMap = new HashMap<>(); + private Map chainSHAMap = new HashMap<>(); //script的SHA1加密值 用于轮询时确定script是否变化 - private Map scriptSHAMap = new HashMap<>(); + private Map scriptSHAMap = new HashMap<>(); /** * 初始化 INSTANCE diff --git a/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/ScriptPollingTask.java b/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/ScriptPollingTask.java index 89c7a91ad..af05f195e 100644 --- a/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/ScriptPollingTask.java +++ b/liteflow-rule-plugin/liteflow-rule-sql/src/main/java/com/yomahub/liteflow/parser/sql/util/ScriptPollingTask.java @@ -1,13 +1,11 @@ package com.yomahub.liteflow.parser.sql.util; -import cn.hutool.core.collection.CollUtil; -import cn.hutool.core.util.ReUtil; import cn.hutool.core.util.StrUtil; -import com.yomahub.liteflow.builder.LiteFlowNodeBuilder; -import com.yomahub.liteflow.enums.NodeTypeEnum; +import cn.hutool.crypto.digest.DigestUtil; import com.yomahub.liteflow.flow.FlowBus; import com.yomahub.liteflow.log.LFLog; import com.yomahub.liteflow.log.LFLoggerManager; +import com.yomahub.liteflow.parser.helper.NodeConvertHelper; import com.yomahub.liteflow.parser.sql.exception.ELSQLException; import com.yomahub.liteflow.parser.sql.vo.SQLParserVO; @@ -33,12 +31,6 @@ public class ScriptPollingTask implements Runnable { private static final String CONCAT_WITH_LANGUAGE_PATTERN = "CONCAT_WS(':',{},{},{},{}) as script_concat"; - private static final String SHA_PATTERN = "SHA1({}) AS SHA1"; - - private static final String SHA_PATTERN_FOR_H2 = "RAWTOHEX(HASH('SHA-1', {})) AS SHA1"; - - private static final String SHA_FIELD_NAME = "SHA1"; - private static final String SCRIPT_KEY_FIELD = "script_concat"; public static Connection conn; @@ -60,6 +52,8 @@ public class ScriptPollingTask implements Runnable { @Override public void run() { + PreparedStatement stmt = null; + ResultSet rs = null; try { String scriptTableName = sqlParserVO.getScriptTableName(); String scriptIdField = sqlParserVO.getScriptIdField(); @@ -70,12 +64,6 @@ public class ScriptPollingTask implements Runnable { String applicationName = sqlParserVO.getApplicationName(); String scriptLanguageField = sqlParserVO.getScriptLanguageField(); - String SHAField = StrUtil.format(SHA_PATTERN, scriptDataField); - //h2数据库计算SHA的函数与MySQL不同 - if(StrUtil.equals(sqlParserVO.getDriverClassName(), "org.h2.Driver")){ - SHAField = StrUtil.format(SHA_PATTERN_FOR_H2, scriptDataField); - } - String KeyField; if (StrUtil.isNotBlank(scriptLanguageField)) { KeyField = StrUtil.format(CONCAT_WITH_LANGUAGE_PATTERN, scriptIdField, scriptTypeField, scriptNameField, scriptLanguageField); @@ -83,46 +71,37 @@ public class ScriptPollingTask implements Runnable { KeyField = StrUtil.format(CONCAT_PATTERN, scriptIdField, scriptTypeField, scriptNameField); } - String sqlCmd = StrUtil.format(SQL_PATTERN, KeyField, SHAField, scriptTableName, scriptApplicationNameField); - PreparedStatement stmt = conn.prepareStatement(sqlCmd, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + String sqlCmd = StrUtil.format(SQL_PATTERN, KeyField, scriptDataField, scriptTableName, scriptApplicationNameField); + stmt = conn.prepareStatement(sqlCmd, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); // 设置游标拉取数量 stmt.setFetchSize(FETCH_SIZE_MAX); stmt.setString(1, applicationName); - ResultSet rs = stmt.executeQuery(); + rs = stmt.executeQuery(); Set newScriptSet = new HashSet<>(); while (rs.next()) { String scriptKey = getStringFromResultSet(rs, SCRIPT_KEY_FIELD); - String newSHA = getStringFromResultSet(rs, SHA_FIELD_NAME); + String newData = getStringFromResultSet(rs, scriptDataField); + String newSHA = DigestUtil.sha1Hex(newData); newScriptSet.add(scriptKey); //如果封装的SHAMap中不存在该script 表示该script为新增 if (!scriptSHAMap.containsKey(scriptKey)) { - //获取新script内容 - NodeSimpleVO scriptVO = convert(scriptKey); - ResultSet newScriptRS = getNewScriptRS(scriptDataField, scriptTableName, scriptIdField, - scriptVO.getNodeId(), scriptApplicationNameField, applicationName); - if(newScriptRS.next()) { - String newScriptData = getStringFromResultSet(newScriptRS, scriptDataField); - //新增script - changeScriptNode(scriptVO, newScriptData); - LOG.info("starting reload flow config... create script={}, new value={},", scriptKey, newScriptData); - } + NodeConvertHelper.NodeSimpleVO scriptVO = NodeConvertHelper.convert(scriptKey); + //新增script + NodeConvertHelper.changeScriptNode(scriptVO, newData); + LOG.info("starting reload flow config... create script={}, new value={},", scriptKey, newData); + //加入到shaMap scriptSHAMap.put(scriptKey, newSHA); } else if (!StrUtil.equals(newSHA, scriptSHAMap.get(scriptKey))) { //SHA值发生变化,表示该script的值已被修改,重新拉取变化的script - //获取新script内容 - NodeSimpleVO scriptVO = convert(scriptKey); - ResultSet newScriptRS = getNewScriptRS(scriptDataField, scriptTableName, scriptIdField, - scriptVO.getNodeId(), scriptApplicationNameField, applicationName); - if(newScriptRS.next()) { - String newScriptData = getStringFromResultSet(newScriptRS, scriptDataField); - //修改script - changeScriptNode(scriptVO, newScriptData); - LOG.info("starting reload flow config... update scriptId={}, new value={},", scriptVO.getNodeId(), newScriptData); - } + NodeConvertHelper.NodeSimpleVO scriptVO = NodeConvertHelper.convert(scriptKey); + //修改script + NodeConvertHelper.changeScriptNode(scriptVO, newData); + LOG.info("starting reload flow config... update scriptId={}, new value={},", scriptVO.getNodeId(), newData); + //修改shaMap scriptSHAMap.put(scriptKey, newSHA); } @@ -140,7 +119,7 @@ public class ScriptPollingTask implements Runnable { while(iterator.hasNext()){ String scriptKey = iterator.next(); if (!newScriptSet.contains(scriptKey)) { - NodeSimpleVO scriptVO = convert(scriptKey); + NodeConvertHelper.NodeSimpleVO scriptVO = NodeConvertHelper.convert(scriptKey); //删除script FlowBus.getNodeMap().remove(scriptVO.getNodeId()); LOG.info("starting reload flow config... delete script={}", scriptKey); @@ -152,25 +131,12 @@ public class ScriptPollingTask implements Runnable { } catch (Exception e) { LOG.error("[Exception during SQL script polling] " + e.getMessage(), e); + } finally { + // 关闭连接 + LiteFlowJdbcUtil.close(null, stmt, rs); } } - private ResultSet getNewScriptRS(String scriptDataField, String scriptTableName, String scriptIdField, - String scriptId, String scriptApplicationNameField, String applicationName) { - ResultSet rs = null; - String sqlCmd = StrUtil.format(NEW_SCRIPT_PATTERN, scriptDataField, scriptTableName, - scriptIdField, scriptApplicationNameField); - try{ - PreparedStatement stmt = conn.prepareStatement(sqlCmd, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); - stmt.setString(1, scriptId); - stmt.setString(2, applicationName); - rs = stmt.executeQuery(); - }catch (Exception e) { - throw new ELSQLException(e.getMessage()); - } - return rs; - } - private String getStringFromResultSet(ResultSet rs, String field) throws SQLException { String data = rs.getString(field); if (StrUtil.isBlank(data)) { @@ -178,92 +144,4 @@ public class ScriptPollingTask implements Runnable { } return data; } - - /*script节点的修改/添加*/ - private void changeScriptNode(NodeSimpleVO nodeSimpleVO, String newValue) { - // 有语言类型 - if (StrUtil.isNotBlank(nodeSimpleVO.getLanguage())) { - LiteFlowNodeBuilder.createScriptNode() - .setId(nodeSimpleVO.getNodeId()) - .setType(NodeTypeEnum.getEnumByCode(nodeSimpleVO.getType())) - .setName(nodeSimpleVO.getName()) - .setScript(newValue) - .setLanguage(nodeSimpleVO.getLanguage()) - .build(); - } - // 没有语言类型 - else { - LiteFlowNodeBuilder.createScriptNode() - .setId(nodeSimpleVO.getNodeId()) - .setType(NodeTypeEnum.getEnumByCode(nodeSimpleVO.getType())) - .setName(nodeSimpleVO.getName()) - .setScript(newValue) - .build(); - } - } - - private NodeSimpleVO convert(String scriptKey){ - List matchItemList = ReUtil.findAllGroup0("(?<=[^:]:)[^:]+|[^:]+(?=:[^:])", scriptKey); - if (CollUtil.isEmpty(matchItemList)) { - return null; - } - NodeSimpleVO nodeSimpleVO = new NodeSimpleVO(); - if (matchItemList.size() > 1) { - nodeSimpleVO.setNodeId(matchItemList.get(0)); - nodeSimpleVO.setType(matchItemList.get(1)); - } - - if (matchItemList.size() > 2) { - nodeSimpleVO.setName(matchItemList.get(2)); - } - - if (matchItemList.size() > 3) { - nodeSimpleVO.setLanguage(matchItemList.get(3)); - } - - return nodeSimpleVO; - } - - class NodeSimpleVO { - - private String nodeId; - - private String type; - - private String name = StrUtil.EMPTY; - - private String language; - - public String getNodeId() { - return nodeId; - } - - public void setNodeId(String nodeId) { - this.nodeId = nodeId; - } - - public String getType() { - return type; - } - - public void setType(String type) { - this.type = type; - } - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - public String getLanguage() { - return language; - } - - public void setLanguage(String language) { - this.language = language; - } - } } diff --git a/liteflow-rule-plugin/liteflow-rule-zk/src/main/java/com/yomahub/liteflow/parser/zk/util/ZkParserHelper.java b/liteflow-rule-plugin/liteflow-rule-zk/src/main/java/com/yomahub/liteflow/parser/zk/util/ZkParserHelper.java index c994b2b56..c70633949 100644 --- a/liteflow-rule-plugin/liteflow-rule-zk/src/main/java/com/yomahub/liteflow/parser/zk/util/ZkParserHelper.java +++ b/liteflow-rule-plugin/liteflow-rule-zk/src/main/java/com/yomahub/liteflow/parser/zk/util/ZkParserHelper.java @@ -10,6 +10,7 @@ import com.yomahub.liteflow.builder.LiteFlowNodeBuilder; import com.yomahub.liteflow.builder.el.LiteFlowChainELBuilder; import com.yomahub.liteflow.enums.NodeTypeEnum; import com.yomahub.liteflow.flow.FlowBus; +import com.yomahub.liteflow.parser.helper.NodeConvertHelper; import com.yomahub.liteflow.parser.zk.exception.ZkException; import com.yomahub.liteflow.parser.zk.vo.ZkParserVO; import org.apache.curator.framework.CuratorFramework; @@ -87,7 +88,7 @@ public class ZkParserHelper { List scriptItemContentList = new ArrayList<>(); for (String scriptNodeValue : scriptNodeValueList) { - NodeSimpleVO nodeSimpleVO = convert(scriptNodeValue); + NodeConvertHelper.NodeSimpleVO nodeSimpleVO = NodeConvertHelper.convert(scriptNodeValue); if (Objects.isNull(nodeSimpleVO)) { throw new ZkException(StrUtil.format("The name of the zk node is invalid:{}", scriptNodeValue)); } @@ -179,12 +180,12 @@ public class ZkParserHelper { .contains(type)) { LOG.info("starting reload flow config... {} path={} value={},", type.name(), path, value); String scriptNodeValue = FileNameUtil.getName(path); - NodeSimpleVO nodeSimpleVO = convert(scriptNodeValue); + NodeConvertHelper.NodeSimpleVO nodeSimpleVO = NodeConvertHelper.convert(scriptNodeValue); // 有语言类型 if (StrUtil.isNotBlank(nodeSimpleVO.getLanguage())) { LiteFlowNodeBuilder.createScriptNode() .setId(nodeSimpleVO.getNodeId()) - .setType(NodeTypeEnum.getEnumByCode(nodeSimpleVO.type)) + .setType(NodeTypeEnum.getEnumByCode(nodeSimpleVO.getType())) .setName(nodeSimpleVO.getName()) .setScript(value) .setLanguage(nodeSimpleVO.getLanguage()) @@ -194,7 +195,7 @@ public class ZkParserHelper { else { LiteFlowNodeBuilder.createScriptNode() .setId(nodeSimpleVO.getNodeId()) - .setType(NodeTypeEnum.getEnumByCode(nodeSimpleVO.type)) + .setType(NodeTypeEnum.getEnumByCode(nodeSimpleVO.getType())) .setName(nodeSimpleVO.getName()) .setScript(value) .build(); @@ -203,80 +204,10 @@ public class ZkParserHelper { else if (CuratorCacheListener.Type.NODE_DELETED.equals(type)) { LOG.info("starting reload flow config... delete path={}", path); String scriptNodeValue = FileNameUtil.getName(path); - NodeSimpleVO nodeSimpleVO = convert(scriptNodeValue); + NodeConvertHelper.NodeSimpleVO nodeSimpleVO = NodeConvertHelper.convert(scriptNodeValue); FlowBus.getNodeMap().remove(nodeSimpleVO.getNodeId()); } }); } } - - public NodeSimpleVO convert(String str) { - // 不需要去理解这串正则，就是一个匹配冒号的 - // 一定得是a:b，或是a:b:c...这种完整类型的字符串的 - List matchItemList = ReUtil.findAllGroup0("(?<=[^:]:)[^:]+|[^:]+(?=:[^:])", str); - if (CollUtil.isEmpty(matchItemList)) { - return null; - } - - NodeSimpleVO nodeSimpleVO = new NodeSimpleVO(); - if (matchItemList.size() > 1) { - nodeSimpleVO.setNodeId(matchItemList.get(0)); - nodeSimpleVO.setType(matchItemList.get(1)); - } - - if (matchItemList.size() > 2) { - nodeSimpleVO.setName(matchItemList.get(2)); - } - - if (matchItemList.size() > 3) { - nodeSimpleVO.setLanguage(matchItemList.get(3)); - } - - return nodeSimpleVO; - } - - private static class NodeSimpleVO { - - private String nodeId; - - private String type; - - private String name = ""; - - private String language; - - public String getNodeId() { - return nodeId; - } - - public void setNodeId(String nodeId) { - this.nodeId = nodeId; - } - - public String getType() { - return type; - } - - public void setType(String type) { - this.type = type; - } - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - public String getLanguage() { - return language; - } - - public void setLanguage(String language) { - this.language = language; - } - - } - }