public/liteFlow
1
0
Fork 0
mirror of https://gitee.com/dromara/liteFlow.git synced 2026-06-10 19:26:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

enhancement #I63C31 zk，etcd支持只刷新改变的部分

Browse Source
This commit is contained in:
everywhere.z
2022-11-28 00:15:55 +08:00
parent b9fe4c78d5
commit 323bc1e080
5 changed files with 57 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
liteflow-core/src/main/java/com/yomahub/liteflow/core/FlowExecutor.java
View File

@@ -161,6 +161,9 @@ public class FlowExecutor {
throw new FlowExecutorNotInitException(errorMsg);
}
}
//执行钩子
FlowInitHook.executeHook();
}
//此方法就是从原有的配置源主动拉取新的进行刷新

32
liteflow-core/src/main/java/com/yomahub/liteflow/core/FlowInitHook.java Normal file
View File

@@ -0,0 +1,32 @@
package com.yomahub.liteflow.core;
import cn.hutool.core.collection.CollUtil;
import java.util.ArrayList;
import java.util.List;
import java.util.function.BooleanSupplier;
/**
* 流程初始化的钩子类，所有的钩子都放在这里
* 目前钩子主要是放一些第三方中间件的规则监听
* 放的钩子要求都是无入参无返回的，所以这里是BooleanSupplier
* @author Bryan.Zhang
* @since 2.9.4
*/
public class FlowInitHook {
private static List<BooleanSupplier> supplierList;
public static void executeHook(){
if (CollUtil.isNotEmpty(supplierList)){
supplierList.forEach(BooleanSupplier::getAsBoolean);
}
}
public static void addHook(BooleanSupplier hookSupplier){
if (CollUtil.isEmpty(supplierList)){
supplierList = new ArrayList<>();
}
supplierList.add(hookSupplier);
}
}

9
liteflow-rule-plugin/liteflow-rule-etcd/src/main/java/com/yomahub/liteflow/parser/etcd/EtcdXmlELParser.java
View File

@@ -4,6 +4,7 @@ import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.bean.copier.CopyOptions;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import com.yomahub.liteflow.core.FlowInitHook;
import com.yomahub.liteflow.parser.el.ClassXmlFlowELParser;
import com.yomahub.liteflow.parser.etcd.exception.EtcdException;
import com.yomahub.liteflow.parser.etcd.util.EtcdParserHelper;
@@ -13,6 +14,7 @@ import com.yomahub.liteflow.property.LiteflowConfigGetter;
import com.yomahub.liteflow.util.JsonUtil;
import java.util.Objects;
import java.util.function.BooleanSupplier;
/**
* Etcd解析器实现，只支持EL形式的XML，不支持其他的形式
@@ -56,7 +58,12 @@ public class EtcdXmlELParser extends ClassXmlFlowELParser {
try {
String content = etcdParserHelper.getContent();
etcdParserHelper.listen();
FlowInitHook.addHook(() -> {
etcdParserHelper.listen();
return true;
});
return content;
} catch (Exception e){
throw new EtcdException(e.getMessage());

8
liteflow-rule-plugin/liteflow-rule-zk/src/main/java/com/yomahub/liteflow/parser/zk/ZkXmlELParser.java
View File

@@ -4,6 +4,7 @@ import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.bean.copier.CopyOptions;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import com.yomahub.liteflow.core.FlowInitHook;
import com.yomahub.liteflow.parser.el.ClassXmlFlowELParser;
import com.yomahub.liteflow.parser.zk.exception.ZkException;
import com.yomahub.liteflow.parser.zk.util.ZkParserHelper;
@@ -55,7 +56,12 @@ public class ZkXmlELParser extends ClassXmlFlowELParser {
public String parseCustom() {
try{
String content = zkParserHelper.getContent();
zkParserHelper.listenZkNode();
FlowInitHook.addHook(() -> {
zkParserHelper.listenZkNode();
return true;
});
return content;
}catch (Exception e){
throw new ZkException(e.getMessage());

12
liteflow-rule-plugin/liteflow-rule-zk/src/main/java/com/yomahub/liteflow/parser/zk/util/ZkParserHelper.java
View File

@@ -123,11 +123,7 @@ public class ZkParserHelper {
//存在这个节点，但是子节点不存在
List<String> chainNameList = client.getChildren().forPath(zkParserVO.getScriptPath());
if (CollUtil.isEmpty(chainNameList)){
return false;
}
return true;
return !CollUtil.isEmpty(chainNameList);
}catch (Exception e){
return false;
}
@@ -143,6 +139,9 @@ public class ZkParserHelper {
cache1.listenable().addListener((type, oldData, data) -> {
String path = data.getPath();
String value = new String(data.getData());
if (StrUtil.isBlank(value)){
return;
}
if (ListUtil.toList(CuratorCacheListener.Type.NODE_CREATED, CuratorCacheListener.Type.NODE_CHANGED).contains(type)){
LOG.info("starting reload flow config... {} path={} value={},", type.name(), path, value);
String chainName = FileNameUtil.getName(path);
@@ -160,6 +159,9 @@ public class ZkParserHelper {
cache2.listenable().addListener((type, oldData, data) -> {
String path = data.getPath();
String value = new String(data.getData());
if (StrUtil.isBlank(value)){
return;
}
if (ListUtil.toList(CuratorCacheListener.Type.NODE_CREATED, CuratorCacheListener.Type.NODE_CHANGED).contains(type)){
LOG.info("starting reload flow config... {} path={} value={},", type.name(), path, value);
String scriptNodeValue = FileNameUtil.getName(path);