基于Java的Kstry业务架构可视化轻量级框架设计源码

/* * * * Copyright (c) 2020-2024, Lykan (jiashuomeng@gmail.com). * * <p> * * Licensed under the Apache License, Version 2.0 (the "License"); * * you may not use this file except in compliance with the License. * * You may obtain a copy of the License at * * <p> * * http://www.apache.org/licenses/LICENSE-2.0 * * <p> * * Unless required by applicable law or agreed to in writing, software * * distributed under the License is distributed on an "AS IS" BASIS, * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * * See the License for the specific language governing permissions and * * limitations under the License. * */ package cn.kstry.framework.core.engine; import cn.kstry.framework.core.bpmn.*; import cn.kstry.framework.core.bpmn.enums.BpmnTypeEnum; import cn.kstry.framework.core.bpmn.enums.IterateStrategyEnum; import cn.kstry.framework.core.bpmn.extend.ElementIterable; import cn.kstry.framework.core.bpmn.impl.BasicElementIterable; import cn.kstry.framework.core.bus.ContextStoryBus; import cn.kstry.framework.core.bus.IterDataItem; import cn.kstry.framework.core.bus.StoryBus; import cn.kstry.framework.core.component.hook.AsyncFlowHook; import cn.kstry.framework.core.component.hook.Hook; import cn.kstry.framework.core.component.limiter.RateLimiterConfig; import cn.kstry.framework.core.component.limiter.strategy.DemotionFailAcquireStrategy; import cn.kstry.framework.core.component.limiter.strategy.FailAcquireStrategy; import cn.kstry.framework.core.container.component.InvokeProperties; import cn.kstry.framework.core.container.component.MethodWrapper; import cn.kstry.framework.core.container.component.TaskServiceDef; import cn.kstry.framework.core.engine.future.FlowTaskSubscriber; import cn.kstry.framework.core.engine.future.RetryFlowTaskSubscriber; import cn.kstry.framework.core.engine.interceptor.SubProcessInterceptorRepository; import cn.kstry.framework.core.engine.interceptor.TaskInterceptorRepository; import cn.kstry.framework.core.engine.thread.FragmentTask; import cn.kstry.framework.core.engine.thread.InvokeMethodThreadLocal; import cn.kstry.framework.core.engine.thread.MonoFlowTask; import cn.kstry.framework.core.exception.BusinessException; import cn.kstry.framework.core.exception.ExceptionEnum; import cn.kstry.framework.core.exception.KstryException; import cn.kstry.framework.core.monitor.DemotionInfo; import cn.kstry.framework.core.monitor.MonitorTracking; import cn.kstry.framework.core.role.Role; import cn.kstry.framework.core.util.AssertUtil; import cn.kstry.framework.core.util.ExceptionUtil; import cn.kstry.framework.core.util.GlobalUtil; import com.google.common.collect.Lists; import com.google.common.collect.Maps; import org.apache.commons.collections.CollectionUtils; import org.apache.commons.lang3.BooleanUtils; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.tuple.ImmutablePair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Mono; import javax.annotation.Nonnull; import java.lang.reflect.Array; import java.time.Duration; import java.util.*; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.IntStream; import java.util.stream.Stream; /** * 流程任务执行核心 * * @author lykan */ public abstract class FlowTaskCore<T> extends BasicTaskCore<T> { private static final Logger LOGGER = LoggerFactory.getLogger(FlowTaskCore.class); public FlowTaskCore(StoryEngineModule engineModule, FlowRegister flowRegister, Role role, StoryBus storyBus) { super(engineModule, flowRegister, storyBus, role, GlobalUtil.getTaskName(flowRegister.getStartElement(), flowRegister.getRequestId())); } protected void doExe(Role role, StoryBus storyBus, FlowRegister flowRegister) { ContextStoryBus csd; Optional<FlowElement> next; for (csd = new ContextStoryBus(storyBus), next = flowRegister.nextElement(csd); next.isPresent(); csd = new ContextStoryBus(storyBus), next = flowRegister.nextElement(csd)) { FlowElement flowElement = next.get(); if (!doInvoke(role, storyBus, flowRegister, flowElement)) { break; } Optional<AsyncFlowHook<List<FlowElement>>> asyncFlowHook = flowRegister.predictNextElement(csd, flowElement); if (asyncFlowHook.isPresent() && BooleanUtils.isTrue(asyncFlowHook.get().openAsync())) { submitAsyncTask(role, storyBus, flowRegister, asyncFlowHook.get()); } } } private boolean doInvoke(Role role, StoryBus storyBus, FlowRegister flowRegister, FlowElement flowElement) { if (flowElement.getElementType() == BpmnTypeEnum.SUB_PROCESS) { SubProcess subProcess = GlobalUtil.transferNotEmpty(flowElement, SubProcess.class); try { InvokeMethodThreadLocal.setCycleTimes(flowRegister.getCycleTimes()); if (!subProcess.getConditionExpression().map(e -> e.condition(storyBus)).orElse(true)) { return true; } } finally { InvokeMethodThreadLocal.clearCycleTimes(); } SubProcessInterceptorRepository subInterceptorRepository = engineModule.getSubInterceptorRepository(); if (!subInterceptorRepository.postBeforeProcessor(storyBus, subProcess.getStartEvent().getId(), flowRegister.getStoryId())) { return true; } subProcessTaskHandler(role, storyBus, flowRegister, subProcess); return false; } if (flowElement.getElementType() != BpmnTypeEnum.SERVICE_TASK) { return true; } Object result; MethodWrapper methodWrapper; TaskServiceDef taskServiceDef; ServiceTask serviceTask = (ServiceTask) flowElement; try { InvokeMethodThreadLocal.whenServiceInvoke(flowRegister, null, serviceTask, storyBus.getBusinessId()); if (!serviceTask.getConditionExpression().map(e -> e.condition(storyBus)).orElse(true)) { flowRegister.getMonitorTracking().expressionTracking(storyBus.getScopeDataOperator(), serviceTask, false); flowRegister.getMonitorTracking().finishTaskTracking(storyBus.getScopeDataOperator(), flowElement, null); return true; } Optional<TaskServiceDef> taskServiceDefOptional = engineModule.getTaskContainer().getTaskServiceDef(serviceTask.getTaskComponent(), serviceTask.getTaskService(), role); if (!taskServiceDefOptional.isPresent() && serviceTask.allowAbsent()) { flowRegister.getMonitorTracking().finishTaskTracking(storyBus.getScopeDataOperator(), flowElement, null); return true; } taskServiceDef = taskServiceDefOptional.orElseThrow(() -> ExceptionUtil.buildException(null, ExceptionEnum.TASK_SERVICE_MATCH_ERROR, ExceptionEnum.TASK_SERVICE_MATCH_ERROR.getDesc() + GlobalUtil.format(" service task identity: {}", serviceTask.identity()))); methodWrapper = taskServiceDef.getMethodWrapper(); flowRegister.getMonitorTracking().getServiceNodeTracking(flowElement).ifPresent(nodeTracking -> { nodeTracking.setThreadId(Thread.currentThread().getName()); nodeTracking.setMethodName(methodWrapper.getMethod().getName()); nodeTracking.setTargetName(taskServiceDef.getTaskComponentTarget().getTarget().getClass().getName()); nodeTracking.setAbility(Optional.ofNullable(methodWrapper.getAbility()).filter(StringUtils::isNotBlank).orElse(null)); }); } finally { I