public/liteFlow
1
0
Fork 0
mirror of https://gitee.com/dromara/liteFlow.git synced 2026-05-14 12:12:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

!45 支持When节点维度配置其独享线程池

Browse Source 
Merge pull request !45 from sikadai/v2.6.9-when-threadpool
This commit is contained in:
铂赛东
2022-01-24 07:16:36 +00:00
committed by Gitee
parent 052f2ce954 b2ef9d6040
commit ab3115e4b5
39 changed files with 785 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
liteflow-core/src/main/java/com/yomahub/liteflow/builder/LiteFlowWhenConditionBuilder.java
View File

@@ -48,4 +48,13 @@ public class LiteFlowWhenConditionBuilder extends LiteFlowConditionBuilder{
}
return setAny(Boolean.parseBoolean(any));
}
public LiteFlowWhenConditionBuilder setThreadExecutorClass(String executorServiceName){
if (StrUtil.isBlank(executorServiceName)) {
return this;
}
this.condition.setThreadExecutorClass(executorServiceName);
return this;
}
}

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

@@ -13,6 +13,7 @@ import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.ReUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists;
import com.yomahub.liteflow.entity.flow.Node;
import com.yomahub.liteflow.enums.FlowParserTypeEnum;
import com.yomahub.liteflow.exception.*;
import com.yomahub.liteflow.parser.*;
@@ -271,6 +272,11 @@ public class FlowExecutor {
this.execute(chainId, param, slotClazz, slotIndex, true);
}
public <T extends Slot> void invoke(String nodeId, Integer slotIndex) throws Exception {
Node node = FlowBus.getNode(nodeId);
node.execute(slotIndex);
}
public DefaultSlot execute(String chainId) throws Exception {
return this.execute(chainId, null, DefaultSlot.class, null, false);
}

9
liteflow-core/src/main/java/com/yomahub/liteflow/entity/data/AbsSlot.java
View File

@@ -127,6 +127,15 @@ public abstract class AbsSlot implements Slot {
}
}
public <T> Queue<T> getPrivateDeliveryQueue(String nodeId){
String privateDKey = PRIVATE_DELIVERY_PREFIX + nodeId;
if(dataMap.containsKey(privateDKey)){
return (Queue<T>) dataMap.get(privateDKey);
}else{
return null;
}
}
public <T> T getPrivateDeliveryData(String nodeId){
String privateDKey = PRIVATE_DELIVERY_PREFIX + nodeId;
if(dataMap.containsKey(privateDKey)){

3
liteflow-core/src/main/java/com/yomahub/liteflow/entity/flow/Chain.java
View File

@@ -10,7 +10,6 @@ package com.yomahub.liteflow.entity.flow;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.ttl.threadpool.TtlExecutors;
import com.yomahub.liteflow.entity.data.DataBus;
import com.yomahub.liteflow.entity.data.Slot;
import com.yomahub.liteflow.entity.flow.parallel.CompletableFutureTimeout;
@@ -120,7 +119,7 @@ public class Chain implements Executable {
Slot slot = DataBus.getSlot(slotIndex);
//此方法其实只会初始化一次Executor，不会每次都会初始化。Executor是唯一的
ExecutorService parallelExecutor = ExecutorHelper.loadInstance().buildExecutor();
ExecutorService parallelExecutor = ExecutorHelper.loadInstance().buildExecutor(condition.getThreadExecutorClass());
//获得liteflow的参数
LiteflowConfig liteflowConfig = LiteflowConfigGetter.get();

10
liteflow-core/src/main/java/com/yomahub/liteflow/entity/flow/Condition.java
View File

@@ -31,6 +31,8 @@ public class Condition {
//只在when类型下有效，为true的话说明在多个并行节点下，任意一个成功，整个when就成功
private boolean any = false;
// when单独的线程池名称
private String threadExecutorClass;
public Condition(List<Executable> nodeList) {
this.nodeList = nodeList;
@@ -77,4 +79,12 @@ public class Condition {
public void setAny(boolean any) {
this.any = any;
}
public String getThreadExecutorClass() {
return threadExecutorClass;
}
public void setThreadExecutorClass(String threadExecutorClass) {
this.threadExecutorClass = threadExecutorClass;
}
}

1
liteflow-core/src/main/java/com/yomahub/liteflow/entity/flow/WhenCondition.java
View File

@@ -19,6 +19,7 @@ public class WhenCondition extends Condition{
super.setGroup(condition.getGroup());
super.setErrorResume(condition.isErrorResume());
super.setAny(condition.isAny());
super.setThreadExecutorClass(condition.getThreadExecutorClass());
}
}

9
liteflow-core/src/main/java/com/yomahub/liteflow/parser/JsonFlowParser.java
View File

@@ -2,7 +2,6 @@ package com.yomahub.liteflow.parser;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.io.resource.ResourceUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONArray;
@@ -11,23 +10,18 @@ import com.alibaba.fastjson.parser.Feature;
import com.yomahub.liteflow.builder.LiteFlowChainBuilder;
import com.yomahub.liteflow.builder.LiteFlowConditionBuilder;
import com.yomahub.liteflow.builder.LiteFlowNodeBuilder;
import com.yomahub.liteflow.common.LocalDefaultFlowConstant;
import com.yomahub.liteflow.core.NodeComponent;
import com.yomahub.liteflow.entity.flow.*;
import com.yomahub.liteflow.enums.ConditionTypeEnum;
import com.yomahub.liteflow.enums.NodeTypeEnum;
import com.yomahub.liteflow.exception.EmptyConditionValueException;
import com.yomahub.liteflow.exception.ExecutableItemNotFoundException;
import com.yomahub.liteflow.exception.NodeTypeNotSupportException;
import com.yomahub.liteflow.exception.NotSupportConditionException;
import com.yomahub.liteflow.flow.FlowBus;
import com.yomahub.liteflow.spring.ComponentScanner;
import org.dom4j.Element;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.function.Consumer;
/**
* Json格式解析器
@@ -136,6 +130,7 @@ public abstract class JsonFlowParser extends FlowParser {
String group;
String errorResume;
String any;
String threadExecutorClass;
//构建chainBuilder
String chainName = chainObject.getString("name");
@@ -148,6 +143,7 @@ public abstract class JsonFlowParser extends FlowParser {
errorResume = condObject.getString("errorResume");
group = condObject.getString("group");
any = condObject.getString("any");
threadExecutorClass = condObject.getString("threadExecutorClass");
if (ObjectUtil.isNull(conditionType)){
throw new NotSupportConditionException("ConditionType is not supported");
@@ -165,6 +161,7 @@ public abstract class JsonFlowParser extends FlowParser {
.setErrorResume(errorResume)
.setGroup(group)
.setAny(any)
.setThreadExecutorClass(threadExecutorClass)
.setValue(condValueStr)
.build()
).build();

11
liteflow-core/src/main/java/com/yomahub/liteflow/parser/XmlFlowParser.java
View File

@@ -2,18 +2,12 @@ package com.yomahub.liteflow.parser;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.io.resource.ResourceUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.yomahub.liteflow.builder.LiteFlowChainBuilder;
import com.yomahub.liteflow.builder.LiteFlowConditionBuilder;
import com.yomahub.liteflow.builder.LiteFlowNodeBuilder;
import com.yomahub.liteflow.common.LocalDefaultFlowConstant;
import com.yomahub.liteflow.core.NodeComponent;
import com.yomahub.liteflow.entity.flow.Chain;
import com.yomahub.liteflow.entity.flow.Condition;
import com.yomahub.liteflow.entity.flow.Executable;
import com.yomahub.liteflow.entity.flow.Node;
import com.yomahub.liteflow.enums.ConditionTypeEnum;
import com.yomahub.liteflow.enums.NodeTypeEnum;
import com.yomahub.liteflow.exception.*;
@@ -25,11 +19,9 @@ import org.dom4j.Element;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.function.Consumer;
/**
* xml形式的解析器
@@ -127,6 +119,7 @@ public abstract class XmlFlowParser extends FlowParser {
String group;
String errorResume;
String any;
String threadExecutorClass;
ConditionTypeEnum conditionType;
//构建chainBuilder
@@ -140,6 +133,7 @@ public abstract class XmlFlowParser extends FlowParser {
errorResume = condE.attributeValue("errorResume");
group = condE.attributeValue("group");
any = condE.attributeValue("any");
threadExecutorClass = condE.attributeValue("threadExecutorClass");
if (ObjectUtil.isNull(conditionType)){
throw new NotSupportConditionException("ConditionType is not supported");
@@ -156,6 +150,7 @@ public abstract class XmlFlowParser extends FlowParser {
.setErrorResume(errorResume)
.setGroup(group)
.setAny(any)
.setThreadExecutorClass(threadExecutorClass)
.setValue(condValueStr)
.build()
).build();

35
liteflow-core/src/main/java/com/yomahub/liteflow/thread/ExecutorBuilder.java
View File

@@ -1,13 +1,48 @@
package com.yomahub.liteflow.thread;
import com.alibaba.ttl.threadpool.TtlExecutors;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
/**
* 并行多线程执行器构造器接口
*
* @author Bryan.Zhang
* @since 2.6.6
*/
public interface ExecutorBuilder {
ExecutorService buildExecutor();
/**
* <p>
* 构建默认的线程池对象
* </p>
* @author sikadai
* @date 2022/1/21 23:07
* @param corePoolSize : 核心线程池数量
* @param maximumPoolSize : 最大线程池数量
* @param queueCapacity : 队列的容量
* @param threadName : 线程吃名称
* @return java.util.concurrent.ExecutorService
*/
default ExecutorService buildDefaultExecutor(int corePoolSize, int maximumPoolSize, int queueCapacity, String threadName) {
return TtlExecutors.getTtlExecutorService(new ThreadPoolExecutor(corePoolSize,
maximumPoolSize,
0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(queueCapacity),
new ThreadFactory() {
private final AtomicLong number = new AtomicLong();
@Override
public Thread newThread(Runnable r) {
Thread newThread = Executors.defaultThreadFactory().newThread(r);
newThread.setName(threadName + number.getAndIncrement());
newThread.setDaemon(false);
return newThread;
}
},
new ThreadPoolExecutor.AbortPolicy()));
}
}

85
liteflow-core/src/main/java/com/yomahub/liteflow/thread/ExecutorHelper.java
View File

@@ -1,6 +1,7 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
*
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
@@ -8,31 +9,44 @@
package com.yomahub.liteflow.thread;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Maps;
import com.yomahub.liteflow.exception.ThreadExecutorServiceCreateException;
import com.yomahub.liteflow.property.LiteflowConfig;
import com.yomahub.liteflow.util.SpringAware;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.concurrent.*;
/**
* 线程池工具类
*
* @author Bryan.Zhang
*/
public class ExecutorHelper {
private final Logger LOG = LoggerFactory.getLogger(ExecutorHelper.class);
private static ExecutorHelper executorHelper;
private ExecutorService executorService;
/**
* 此处使用Map缓存线程池信息
* key - 线程池构建者的Class全类名
* value - 线程池对象
*/
private final Map<String, ExecutorService> executorServiceMap;
private ExecutorHelper() {
executorServiceMap = Maps.newConcurrentMap();
}
public static ExecutorHelper loadInstance(){
if (ObjectUtil.isNull(executorHelper)){
public static ExecutorHelper loadInstance() {
if (ObjectUtil.isNull(executorHelper)) {
executorHelper = new ExecutorHelper();
}
return executorHelper;
@@ -56,7 +70,7 @@ public class ExecutorHelper {
* @param timeout 等待时间
*/
public void shutdownAwaitTermination(ExecutorService pool,
long timeout) {
long timeout) {
pool.shutdown();
try {
if (!pool.awaitTermination(timeout, TimeUnit.SECONDS)) {
@@ -71,23 +85,62 @@ public class ExecutorHelper {
}
}
/**
* 构建全局默认线程池
*/
public ExecutorService buildExecutor() {
if (ObjectUtil.isNull(executorService)){
if (ObjectUtil.isNull(executorService)) {
LiteflowConfig liteflowConfig = SpringAware.getBean(LiteflowConfig.class);
try{
assert liteflowConfig != null;
ExecutorBuilder executorBuilder = (ExecutorBuilder)Class.forName(liteflowConfig.getThreadExecutorClass()).newInstance();
executorService = executorBuilder.buildExecutor();
}catch (Exception e){
LOG.error(e.getMessage(), e);
throw new ThreadExecutorServiceCreateException(e.getMessage());
}
assert liteflowConfig != null;
executorService = getExecutorBuilder(liteflowConfig.getThreadExecutorClass()).buildExecutor();
}
return executorService;
}
/**
* <p>
* 构建线程池执行器 - 支持多个when公用一个线程池
* </p>
*
* @param threadExecutorClass : 线程池构建者的Class全类名
* @return java.util.concurrent.ExecutorService
* @author sikadai
* @date 2022/1/21 23:00
*/
public ExecutorService buildExecutor(String threadExecutorClass) {
if (StrUtil.isBlank(threadExecutorClass)) {
return buildExecutor();
}
ExecutorService executorServiceFromCache = executorServiceMap.get(threadExecutorClass);
if (executorServiceFromCache != null) {
return executorServiceFromCache;
} else {
ExecutorService executorService = getExecutorBuilder(threadExecutorClass).buildExecutor();
executorServiceMap.put(threadExecutorClass, executorService);
return executorService;
}
}
/**
* <p>
* 根据线程执行构建者Class类名获取ExecutorBuilder实例
* </p>
*
* @param threadExecutorClass
* @return com.yomahub.liteflow.thread.ExecutorBuilder
* @author sikadai
* @date 2022/1/21 23:04
*/
private ExecutorBuilder getExecutorBuilder(String threadExecutorClass) {
try {
return (ExecutorBuilder) Class.forName(threadExecutorClass).newInstance();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
throw new ThreadExecutorServiceCreateException(e.getMessage());
}
}
public ExecutorService getExecutorService() {
return executorService;
}

19
liteflow-core/src/main/java/com/yomahub/liteflow/thread/LiteFlowDefaultExecutorBuilder.java
View File

@@ -21,21 +21,10 @@ public class LiteFlowDefaultExecutorBuilder implements ExecutorBuilder{
if (ObjectUtil.isNull(liteflowConfig)){
liteflowConfig = new LiteflowConfig();
}
return TtlExecutors.getTtlExecutorService(new ThreadPoolExecutor(liteflowConfig.getWhenMaxWorkers(),
return buildDefaultExecutor(
liteflowConfig.getWhenMaxWorkers(),
0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(liteflowConfig.getWhenQueueLimit()),
new ThreadFactory() {
private final AtomicLong number = new AtomicLong();
@Override
public Thread newThread(Runnable r) {
Thread newThread = Executors.defaultThreadFactory().newThread(r);
newThread.setName("lf-when-thead-" + number.getAndIncrement());
newThread.setDaemon(false);
return newThread;
}
},
new ThreadPoolExecutor.AbortPolicy()));
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenQueueLimit(),
"lf-when-thead-");
}
}

11
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/builder/BuilderSpringbootTest.java
View File

@@ -32,7 +32,7 @@ public class BuilderSpringbootTest extends BaseTest {
//基于普通组件的builder模式测试
@Test
public void testBuilder() throws Exception{
public void testBuilder() throws Exception {
LiteFlowNodeBuilder.createNode().setId("a")
.setName("组件A")
.setType(NodeTypeEnum.COMMON)
@@ -71,13 +71,16 @@ public class BuilderSpringbootTest extends BaseTest {
LiteFlowChainBuilder.createChain().setChainName("chain2").setCondition(
LiteFlowConditionBuilder.createThenCondition().setValue("c,d").build()
LiteFlowConditionBuilder.createWhenCondition().setValue("c,d").build()
).build();
LiteFlowChainBuilder.createChain().setChainName("chain1").setCondition(
LiteFlowConditionBuilder.createThenCondition().setValue("a,b").build()
LiteFlowConditionBuilder
.createWhenCondition()
.setValue("a,b").build()
).setCondition(
LiteFlowConditionBuilder.createWhenCondition().setValue("e(f|g|chain2)").build()
LiteFlowConditionBuilder.createWhenCondition()
.setValue("e(f|g|chain2)").build()
).build();
LiteflowResponse<DefaultSlot> response = flowExecutor.execute2Resp("chain1");

27
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/CustomThreadExecutor1.java Normal file
View File

@@ -0,0 +1,27 @@
package com.yomahub.liteflow.test.customWhenThreadPool;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.ttl.threadpool.TtlExecutors;
import com.yomahub.liteflow.property.LiteflowConfig;
import com.yomahub.liteflow.thread.ExecutorBuilder;
import com.yomahub.liteflow.util.SpringAware;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
public class CustomThreadExecutor1 implements ExecutorBuilder {
@Override
public ExecutorService buildExecutor() {
LiteflowConfig liteflowConfig = SpringAware.getBean(LiteflowConfig.class);
//只有在非spring的场景下liteflowConfig才会为null
if (ObjectUtil.isNull(liteflowConfig)) {
liteflowConfig = new LiteflowConfig();
}
return buildDefaultExecutor(
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenQueueLimit(),
"customer-when-1-thead-");
}
}

26
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/CustomThreadExecutor2.java Normal file
View File

@@ -0,0 +1,26 @@
package com.yomahub.liteflow.test.customWhenThreadPool;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.ttl.threadpool.TtlExecutors;
import com.yomahub.liteflow.property.LiteflowConfig;
import com.yomahub.liteflow.thread.ExecutorBuilder;
import com.yomahub.liteflow.util.SpringAware;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
public class CustomThreadExecutor2 implements ExecutorBuilder {
@Override
public ExecutorService buildExecutor() {
LiteflowConfig liteflowConfig = SpringAware.getBean(LiteflowConfig.class);
//只有在非spring的场景下liteflowConfig才会为null
if (ObjectUtil.isNull(liteflowConfig)) {
liteflowConfig = new LiteflowConfig();
}
return buildDefaultExecutor(
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenQueueLimit(),
"customer-when-2-thead-");
}
}

24
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/CustomThreadExecutor3.java Normal file
View File

@@ -0,0 +1,24 @@
package com.yomahub.liteflow.test.customWhenThreadPool;
import cn.hutool.core.util.ObjectUtil;
import com.yomahub.liteflow.property.LiteflowConfig;
import com.yomahub.liteflow.thread.ExecutorBuilder;
import com.yomahub.liteflow.util.SpringAware;
import java.util.concurrent.ExecutorService;
public class CustomThreadExecutor3 implements ExecutorBuilder {
@Override
public ExecutorService buildExecutor() {
LiteflowConfig liteflowConfig = SpringAware.getBean(LiteflowConfig.class);
//只有在非spring的场景下liteflowConfig才会为null
if (ObjectUtil.isNull(liteflowConfig)) {
liteflowConfig = new LiteflowConfig();
}
return buildDefaultExecutor(
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenQueueLimit(),
"customer-when-3-thead-");
}
}

72
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/CustomWhenThreadPoolSpringbootTest.java Normal file
View File

@@ -0,0 +1,72 @@
package com.yomahub.liteflow.test.customWhenThreadPool;
import com.yomahub.liteflow.core.FlowExecutor;
import com.yomahub.liteflow.entity.data.DefaultSlot;
import com.yomahub.liteflow.entity.data.LiteflowResponse;
import com.yomahub.liteflow.test.BaseTest;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;
/**
* springboot环境下异步线程超时日志打印测试
*
* @author Bryan.Zhang
* @since 2.6.4
*/
@RunWith(SpringRunner.class)
@TestPropertySource(value = "classpath:/customWhenThreadPool/application.properties")
@SpringBootTest(classes = CustomWhenThreadPoolSpringbootTest.class)
@EnableAutoConfiguration
@ComponentScan({"com.yomahub.liteflow.test.customWhenThreadPool.cmp"})
public class CustomWhenThreadPoolSpringbootTest extends BaseTest {
private final Logger log = LoggerFactory.getLogger(this.getClass());
@Resource
private FlowExecutor flowExecutor;
/**
* 测试全局线程池配置
*/
@Test
public void testGlobalThreadPool() {
LiteflowResponse<DefaultSlot> response = flowExecutor.execute2Resp("chain", "arg");
Assert.assertTrue(response.isSuccess());
Assert.assertTrue(response.getSlot().getData("threadName").toString().startsWith("lf-when-thead"));
}
/**
* 测试全局和when上自定义线程池-优先以when上为准
*/
@Test
public void testGlobalAndCustomWhenThreadPool() {
LiteflowResponse<DefaultSlot> response1 = flowExecutor.execute2Resp("chain1", "arg");
Assert.assertTrue(response1.isSuccess());
Assert.assertTrue(response1.getSlot().getData("threadName").toString().startsWith("customer-when-1-thead"));
}
/**
* when配置的线程池可以共用
*/
@Test
public void testCustomWhenThreadPool() {
// 使用when - thread1
testGlobalAndCustomWhenThreadPool();
// chain配置同一个thead1
LiteflowResponse<DefaultSlot> response2 = flowExecutor.execute2Resp("chain2", "arg");
Assert.assertTrue(response2.isSuccess());
Assert.assertTrue(response2.getSlot().getData("threadName").toString().startsWith("customer-when-1-thead"));
}
}

20
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/ACmp.java Normal file
View File

@@ -0,0 +1,20 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("a")
public class ACmp extends NodeComponent {
@Override
public void process() {
System.out.println("ACmp executed!");
}
}

22
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/BCmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("b")
public class BCmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("BCmp executed!");
}
}

22
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/CCmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("c")
public class CCmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("CCmp executed!");
}
}

22
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/DCmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("d")
public class DCmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("DCmp executed!");
}
}

22
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/ECmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("e")
public class ECmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("ECmp executed!");
}
}

22
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/FCmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("f")
public class FCmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("FCmp executed!");
}
}

2
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/privateDelivery/cmp/ACmp.java
View File

@@ -14,6 +14,7 @@ import org.springframework.stereotype.Component;
import java.util.HashSet;
@Component("a")
@LiteflowComponent("a")
public class ACmp extends NodeComponent {
@@ -28,3 +29,4 @@ public class ACmp extends NodeComponent {
}
}
}

16
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/privateDelivery/cmp/BCmp.java
View File

@@ -13,14 +13,16 @@ import org.springframework.stereotype.Component;
import java.util.Set;
@Component("b")
@LiteflowComponent("b")
public class BCmp extends NodeComponent {
@Override
public void process() {
System.out.println("BCmp executed!");
Integer value = this.getPrivateDeliveryData();
Set<Integer> testSet = this.getSlot().getData("testSet");
testSet.add(value);
}
@Override
public void process() {
System.out.println("BCmp executed!");
Integer value = this.getPrivateDeliveryData();
Set<Integer> testSet = this.getSlot().getData("testSet");
testSet.add(value);
}
}

21
liteflow-testcase-springboot/src/test/java/com/yomahub/liteflow/test/privateDelivery/cmp/DCmp.java Normal file
View File

@@ -0,0 +1,21 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.privateDelivery.cmp;
import com.yomahub.liteflow.annotation.LiteflowComponent;
import com.yomahub.liteflow.core.NodeComponent;
@LiteflowComponent("d")
public class DCmp extends NodeComponent {
@Override
public void process() {
System.out.println("CCmp executed!");
}
}

1
liteflow-testcase-springboot/src/test/resources/customWhenThreadPool/application.properties Normal file
View File

@@ -0,0 +1 @@
liteflow.rule-source=customWhenThreadPool/flow.xml

12
liteflow-testcase-springboot/src/test/resources/customWhenThreadPool/flow.xml Normal file
View File

@@ -0,0 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<flow>
<chain name="chain">
<when value="a,b"/>
</chain>
<chain name="chain1">
<when value="c,d" threadExecutorClass="com.yomahub.liteflow.test.customWhenThreadPool.CustomThreadExecutor1"/>
</chain>
<chain name="chain2">
<when value="e,f" threadExecutorClass="com.yomahub.liteflow.test.customWhenThreadPool.CustomThreadExecutor1"/>
</chain>
</flow>

25
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/CustomThreadExecutor1.java Normal file
View File

@@ -0,0 +1,25 @@
package com.yomahub.liteflow.test.customWhenThreadPool;
import cn.hutool.core.util.ObjectUtil;
import com.yomahub.liteflow.property.LiteflowConfig;
import com.yomahub.liteflow.thread.ExecutorBuilder;
import com.yomahub.liteflow.util.SpringAware;
import java.util.concurrent.ExecutorService;
public class CustomThreadExecutor1 implements ExecutorBuilder {
@Override
public ExecutorService buildExecutor() {
LiteflowConfig liteflowConfig = SpringAware.getBean(LiteflowConfig.class);
//只有在非spring的场景下liteflowConfig才会为null
if (ObjectUtil.isNull(liteflowConfig)) {
liteflowConfig = new LiteflowConfig();
}
return buildDefaultExecutor(
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenQueueLimit(),
"customer-when-1-thead-");
}
}

24
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/CustomThreadExecutor2.java Normal file
View File

@@ -0,0 +1,24 @@
package com.yomahub.liteflow.test.customWhenThreadPool;
import cn.hutool.core.util.ObjectUtil;
import com.yomahub.liteflow.property.LiteflowConfig;
import com.yomahub.liteflow.thread.ExecutorBuilder;
import com.yomahub.liteflow.util.SpringAware;
import java.util.concurrent.ExecutorService;
public class CustomThreadExecutor2 implements ExecutorBuilder {
@Override
public ExecutorService buildExecutor() {
LiteflowConfig liteflowConfig = SpringAware.getBean(LiteflowConfig.class);
//只有在非spring的场景下liteflowConfig才会为null
if (ObjectUtil.isNull(liteflowConfig)) {
liteflowConfig = new LiteflowConfig();
}
return buildDefaultExecutor(
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenQueueLimit(),
"customer-when-2-thead-");
}
}

24
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/CustomThreadExecutor3.java Normal file
View File

@@ -0,0 +1,24 @@
package com.yomahub.liteflow.test.customWhenThreadPool;
import cn.hutool.core.util.ObjectUtil;
import com.yomahub.liteflow.property.LiteflowConfig;
import com.yomahub.liteflow.thread.ExecutorBuilder;
import com.yomahub.liteflow.util.SpringAware;
import java.util.concurrent.ExecutorService;
public class CustomThreadExecutor3 implements ExecutorBuilder {
@Override
public ExecutorService buildExecutor() {
LiteflowConfig liteflowConfig = SpringAware.getBean(LiteflowConfig.class);
//只有在非spring的场景下liteflowConfig才会为null
if (ObjectUtil.isNull(liteflowConfig)) {
liteflowConfig = new LiteflowConfig();
}
return buildDefaultExecutor(
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenMaxWorkers(),
liteflowConfig.getWhenQueueLimit(),
"customer-when-3-thead-");
}
}

66
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/CustomWhenThreadPoolSpringTest.java Normal file
View File

@@ -0,0 +1,66 @@
package com.yomahub.liteflow.test.customWhenThreadPool;
import com.yomahub.liteflow.core.FlowExecutor;
import com.yomahub.liteflow.entity.data.DefaultSlot;
import com.yomahub.liteflow.entity.data.LiteflowResponse;
import com.yomahub.liteflow.test.BaseTest;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;
/**
* springboot环境下异步线程超时日志打印测试
*
* @author Bryan.Zhang
* @since 2.6.4
*/
@RunWith(SpringRunner.class)
@ContextConfiguration("classpath:/customWhenThreadPool/application.xml")
public class CustomWhenThreadPoolSpringTest extends BaseTest {
private final Logger log = LoggerFactory.getLogger(this.getClass());
@Resource
private FlowExecutor flowExecutor;
/**
* 测试全局线程池配置
*/
@Test
public void testGlobalThreadPool() {
LiteflowResponse<DefaultSlot> response = flowExecutor.execute2Resp("chain", "arg");
Assert.assertTrue(response.isSuccess());
Assert.assertTrue(response.getSlot().getData("threadName").toString().startsWith("lf-when-thead"));
}
/**
* 测试全局和when上自定义线程池-优先以when上为准
*/
@Test
public void testGlobalAndCustomWhenThreadPool() {
LiteflowResponse<DefaultSlot> response1 = flowExecutor.execute2Resp("chain1", "arg");
Assert.assertTrue(response1.isSuccess());
Assert.assertTrue(response1.getSlot().getData("threadName").toString().startsWith("customer-when-1-thead"));
}
/**
* when配置的线程池可以共用
*/
@Test
public void testCustomWhenThreadPool() {
// 使用when - thread1
testGlobalAndCustomWhenThreadPool();
// chain配置同一个thead1
LiteflowResponse<DefaultSlot> response2 = flowExecutor.execute2Resp("chain2", "arg");
Assert.assertTrue(response2.isSuccess());
Assert.assertTrue(response2.getSlot().getData("threadName").toString().startsWith("customer-when-1-thead"));
}
}

20
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/ACmp.java Normal file
View File

@@ -0,0 +1,20 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("a")
public class ACmp extends NodeComponent {
@Override
public void process() {
System.out.println("ACmp executed!");
}
}

22
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/BCmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("b")
public class BCmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("BCmp executed!");
}
}

22
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/CCmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("c")
public class CCmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("CCmp executed!");
}
}

22
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/DCmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("d")
public class DCmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("DCmp executed!");
}
}

22
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/ECmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("e")
public class ECmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("ECmp executed!");
}
}

22
liteflow-testcase-springnative/src/test/java/com/yomahub/liteflow/test/customWhenThreadPool/cmp/FCmp.java Normal file
View File

@@ -0,0 +1,22 @@
/**
* <p>Title: liteflow</p>
* <p>Description: 轻量级的组件式流程框架</p>
* @author Bryan.Zhang
* @email weenyc31@163.com
* @Date 2020/4/1
*/
package com.yomahub.liteflow.test.customWhenThreadPool.cmp;
import com.yomahub.liteflow.core.NodeComponent;
import org.springframework.stereotype.Component;
@Component("f")
public class FCmp extends NodeComponent {
@Override
public void process() {
this.getSlot().setData("threadName", Thread.currentThread().getName());
System.out.println("FCmp executed!");
}
}

23
liteflow-testcase-springnative/src/test/resources/customWhenThreadPool/application.xml Normal file
View File

@@ -0,0 +1,23 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://www.springframework.org/schema/beans"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.0.xsd">
<context:component-scan base-package="com.yomahub.liteflow.test.customWhenThreadPool.cmp" />
<bean id="springAware" class="com.yomahub.liteflow.util.SpringAware"/>
<bean class="com.yomahub.liteflow.spring.ComponentScanner"/>
<bean id="liteflowConfig" class="com.yomahub.liteflow.property.LiteflowConfig">
<property name="ruleSource" value="customWhenThreadPool/flow.xml"/>
</bean>
<bean id="flowExecutor" class="com.yomahub.liteflow.core.FlowExecutor">
<property name="liteflowConfig" ref="liteflowConfig"/>
</bean>
</beans>

12
liteflow-testcase-springnative/src/test/resources/customWhenThreadPool/flow.xml Normal file
View File

@@ -0,0 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<flow>
<chain name="chain">
<when value="a,b"/>
</chain>
<chain name="chain1">
<when value="c,d" threadExecutorClass="com.yomahub.liteflow.test.customWhenThreadPool.CustomThreadExecutor1"/>
</chain>
<chain name="chain2">
<when value="e,f" threadExecutorClass="com.yomahub.liteflow.test.customWhenThreadPool.CustomThreadExecutor1"/>
</chain>
</flow>