wangli/workflow-engine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'refs/heads/hotfix/improve-batch-operation' into feature/starter

Browse Source 
# Conflicts:
#	workflow-engine-server/src/main/java/cn/axzo/workflow/server/common/aspectj/ErrorReportAspect.java
This commit is contained in:
wangli 2024-06-28 10:48:33 +08:00
commit 05981a7002
15 changed files with 133 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
workflow-engine-core/src/main/java/cn/axzo/workflow/core/common/event/ApiLogListener.java
View File

@ -5,15 +5,20 @@ import cn.axzo.workflow.core.repository.entity.ExtAxApiLog;
import cn.axzo.workflow.core.service.ExtAxApiLogService;
import cn.hutool.json.JSONUtil;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
@ -45,6 +50,9 @@ public class ApiLogListener implements ApplicationListener<ApiLogEvent> {
if (!refreshProperties.getApiLogEnable() || EXCLUDE_URLS.contains(event.getApiUrl())) {
return;
}
if (!supportApiType(event)) {
return;
}
rebuildTraceId(event);
ExtAxApiLog apiLog = new ExtAxApiLog();
@ -65,4 +73,12 @@ public class ApiLogListener implements ApplicationListener<ApiLogEvent> {
event.setTraceId(traceId);
}
}
private boolean supportApiType(ApiLogEvent event) {
if (!StringUtils.hasText(refreshProperties.getFilterApiType())) {
return true;
}
List<String> list = Arrays.asList(refreshProperties.getFilterApiType().split(","));
return list.contains(event.getType());
}
}

9
workflow-engine-core/src/main/java/cn/axzo/workflow/core/conf/SupportRefreshProperties.java
View File

@ -1,10 +1,14 @@
package cn.axzo.workflow.core.conf;
import com.google.common.collect.Lists;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
/**
* 支持动态刷新配置属性
*
@ -19,6 +23,9 @@ public class SupportRefreshProperties {
@Value("${workflow.apiLog.enable: false}")
private Boolean apiLogEnable;
@Value("${workflow.mqLog.enable: true}")
@Value("${workflow.mqLog.enable: false}")
private Boolean mqLogEnable;
@Value("${workflow.apiLog.filterApiType:}")
private String filterApiType;
}

3
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/cmd/helper/CustomTaskHelper.java
View File

@ -9,6 +9,7 @@ import cn.axzo.workflow.core.common.utils.BpmnMetaParserHelper;
import cn.axzo.workflow.core.repository.entity.ExtAxHiTaskInst;
import cn.axzo.workflow.core.service.ExtAxHiTaskInstService;
import cn.axzo.workflow.core.service.converter.BpmnHistoricTaskInstanceConverter;
import lombok.extern.slf4j.Slf4j;
import org.flowable.bpmn.model.FlowElement;
import org.flowable.bpmn.model.Process;
import org.flowable.common.engine.impl.cfg.IdGenerator;
@ -69,6 +70,7 @@ import static org.flowable.task.api.Task.DEFAULT_PRIORITY;
* @author wangli
* @since 2023/12/22 10:59
*/
@Slf4j
public class CustomTaskHelper {
public static void addMultiTask(CommandContext commandContext, TaskEntity originTask,
@ -150,6 +152,7 @@ public class CustomTaskHelper {
BpmnFlowNodeType nodeType = BpmnMetaParserHelper.getNodeType(currentFlowElement).orElse(NODE_EMPTY);
//不包含对应的任务
if (!nodeTypes.contains(nodeType)) {
// log.warn(TASK_TYPE_MISMATCH.getMessage(), nodeType.getDesc(), nodeTypes.stream().map(BpmnFlowNodeType::getDesc).collect(Collectors.joining(",")));
throw new WorkflowEngineException(TASK_TYPE_MISMATCH, nodeType.getDesc(), nodeTypes.stream().map(BpmnFlowNodeType::getDesc).collect(Collectors.joining(",")));
}
}

50
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/AbstractJobHandler.java Normal file
View File

@ -0,0 +1,50 @@
package cn.axzo.workflow.core.engine.job;
import org.flowable.job.service.JobHandler;
import org.flowable.job.service.impl.persistence.entity.JobEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Collections;
import java.util.Enumeration;
import java.util.List;
/**
* 记录每个 JOB 是否被多台机器同时处理
*
* @author wangli
* @since 2024/6/28 10:28
*/
public abstract class AbstractJobHandler implements JobHandler {
private static final Logger log = LoggerFactory.getLogger(AbstractJobHandler.class);
protected final void log(JobEntity job) {
try {
// 获取所有网络接口
Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
while (interfaces.hasMoreElements()) {
NetworkInterface iface = interfaces.nextElement();
// 如果是活动的，并且不是回环接口（即非本地接口）
if (iface.isUp() && !iface.isLoopback()) {
List<InetAddress> addresses = Collections.list(iface.getInetAddresses());
// 遍历每个接口上的 IP 地址
for (InetAddress address : addresses) {
// 检查地址是否是 IPv4 类型的，并且不是回环地址
if (address instanceof Inet4Address && !address.isLoopbackAddress()) {
log.warn(" current job ID: {} will be execution by IP:{}", job.getId(), address.getHostAddress());
}
}
}
}
} catch (SocketException e) {
// ignore
}
}
}

4
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/AsyncAbortProcessInstanceHandler.java
View File

@ -13,7 +13,7 @@ import org.flowable.job.service.impl.persistence.entity.JobEntity;
import org.flowable.variable.api.delegate.VariableScope;
@Slf4j
public class AsyncAbortProcessInstanceHandler implements JobHandler {
public class AsyncAbortProcessInstanceHandler extends AbstractJobHandler implements JobHandler {
public static final String TYPE = "async-abort-instance";
@ -31,9 +31,11 @@ public class AsyncAbortProcessInstanceHandler implements JobHandler {
@Override
public void execute(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext) {
log.info("AsyncAbortProcessInstanceHandler executing...,jobInfo:{}", JSONUtil.toJsonStr(job));
log(job);
ProcessEngineConfigurationImpl processEngineConfiguration = CommandContextUtil.getProcessEngineConfiguration(commandContext);
BpmnProcessInstanceCancelDTO dto = JSONUtil.toBean(job.getCustomValues(), BpmnProcessInstanceCancelDTO.class);
processEngineConfiguration.getCommandExecutor().execute(new CustomAbortProcessInstanceCmd(dto.getProcessInstanceId(), dto.getTenantId(),
dto.getReason(), extAxHiTaskInstService));
}
}

14
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/AsyncApproveTaskJobHandler.java
View File

@ -12,6 +12,13 @@ import org.flowable.job.service.impl.persistence.entity.JobEntity;
import org.flowable.task.api.Task;
import org.flowable.variable.api.delegate.VariableScope;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Collections;
import java.util.Enumeration;
import java.util.List;
import java.util.Objects;
import static cn.axzo.workflow.common.enums.BpmnFlowNodeType.NODE_STARTER;
@ -23,7 +30,7 @@ import static cn.axzo.workflow.common.enums.BpmnFlowNodeType.NODE_STARTER;
* @since 2024/4/15 22:41
*/
@Slf4j
public class AsyncApproveTaskJobHandler implements JobHandler {
public class AsyncApproveTaskJobHandler extends AbstractJobHandler implements JobHandler {
public static final String TYPE = "async-approve-task";
@Override
@ -34,14 +41,19 @@ public class AsyncApproveTaskJobHandler implements JobHandler {
@Override
public void execute(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext) {
log.info("AsyncApproveTaskJobHandler executing...");
log(job);
ProcessEngineConfigurationImpl processEngineConfiguration =
CommandContextUtil.getProcessEngineConfiguration(commandContext);
BpmnTaskAuditDTO dto = JSONUtil.toBean(job.getCustomValues(), BpmnTaskAuditDTO.class);
Task task = processEngineConfiguration.getTaskService().createTaskQuery().taskId(dto.getTaskId()).singleResult();
if (Objects.isNull(task)) {
return;
}
CustomApproveTaskCmd command = new CustomApproveTaskCmd(dto);
if (Objects.equals(task.getTaskDefinitionKey(), NODE_STARTER.getType())) {
command = new CustomApproveTaskCmd(dto, "");
}
processEngineConfiguration.getCommandExecutor().execute(command);
}
}

3
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/AsyncBpmnProcessActivityJobHandler.java
View File

@ -8,7 +8,7 @@ import org.flowable.job.service.impl.persistence.entity.JobEntity;
import org.flowable.variable.api.delegate.VariableScope;
@Slf4j
public class AsyncBpmnProcessActivityJobHandler implements JobHandler {
public class AsyncBpmnProcessActivityJobHandler extends AbstractJobHandler implements JobHandler {
public static final String TYPE = "async-bpmn-process-activity";
@ -25,6 +25,7 @@ public class AsyncBpmnProcessActivityJobHandler implements JobHandler {
@Override
public void execute(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext) {
log(job);
activityService.executeAsyncJob(job);
}
}

3
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/AsyncCancelProcessInstanceHandler.java
View File

@ -13,7 +13,7 @@ import org.flowable.job.service.impl.persistence.entity.JobEntity;
import org.flowable.variable.api.delegate.VariableScope;
@Slf4j
public class AsyncCancelProcessInstanceHandler implements JobHandler {
public class AsyncCancelProcessInstanceHandler extends AbstractJobHandler implements JobHandler {
public static final String TYPE = "async-cancel-process";
@ -31,6 +31,7 @@ public class AsyncCancelProcessInstanceHandler implements JobHandler {
@Override
public void execute(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext) {
log.info("AsyncCancelProcessInstanceHandler executing...,jobInfo:{}", JSONUtil.toJsonStr(job));
log(job);
ProcessEngineConfigurationImpl processEngineConfiguration = CommandContextUtil.getProcessEngineConfiguration(commandContext);
BpmnProcessInstanceCancelDTO dto = JSONUtil.toBean(job.getCustomValues(), BpmnProcessInstanceCancelDTO.class);
processEngineConfiguration.getCommandExecutor().execute(new CustomCancelProcessInstanceCmd(dto.getProcessInstanceId(), dto.getTenantId(),

3
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/AsyncCountersignUserTaskJobHandler.java
View File

@ -14,7 +14,7 @@ import org.flowable.job.service.impl.persistence.entity.JobEntity;
import org.flowable.variable.api.delegate.VariableScope;
@Slf4j
public class AsyncCountersignUserTaskJobHandler implements JobHandler {
public class AsyncCountersignUserTaskJobHandler extends AbstractJobHandler implements JobHandler {
public static final String TYPE = "async-countersign-task";
@ -32,6 +32,7 @@ public class AsyncCountersignUserTaskJobHandler implements JobHandler {
@Override
public void execute(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext) {
log.info("AsyncCountersignUserTaskJobHandler executing...");
log(job);
ProcessEngineConfigurationImpl processEngineConfiguration =
CommandContextUtil.getProcessEngineConfiguration(commandContext);
BpmnTaskCountersignDTO dto = JSONUtil.toBean(job.getCustomValues(), BpmnTaskCountersignDTO.class);

3
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/AsyncExtTaskInstJobHandler.java
View File

@ -23,7 +23,7 @@ import static cn.axzo.workflow.core.common.code.AsyncJobRespCode.DATA_NOT_EXISTS
* @since 2024/4/29 20:22
*/
@Slf4j
public class AsyncExtTaskInstJobHandler implements JobHandler {
public class AsyncExtTaskInstJobHandler extends AbstractJobHandler implements JobHandler {
public static final String TYPE = "async-update-ext-task-inst";
private final ExtAxHiTaskInstService extAxHiTaskInstService;
@ -38,6 +38,7 @@ public class AsyncExtTaskInstJobHandler implements JobHandler {
@Override
public void execute(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext) {
log(job);
ExtTaskInstUpdateEvent event = JSONObject.parseObject(job.getCustomValues(), ExtTaskInstUpdateEvent.class);
String taskId = event.getTaskId();
String processInstanceId = event.getProcessInstanceId();

3
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/AsyncRejectTaskJobHandler.java
View File

@ -19,7 +19,7 @@ import org.flowable.variable.api.delegate.VariableScope;
* @since 2024/4/16 11:11
*/
@Slf4j
public class AsyncRejectTaskJobHandler implements JobHandler {
public class AsyncRejectTaskJobHandler extends AbstractJobHandler implements JobHandler {
public static final String TYPE = "async-reject-task";
private final ExtAxHiTaskInstService extAxHiTaskInstService;
@ -35,6 +35,7 @@ public class AsyncRejectTaskJobHandler implements JobHandler {
@Override
public void execute(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext) {
log.info("AsyncRejectTaskJobHandler executing...");
log(job);
ProcessEngineConfigurationImpl processEngineConfiguration =
CommandContextUtil.getProcessEngineConfiguration(commandContext);
BpmnTaskAuditDTO dto = JSONUtil.toBean(job.getCustomValues(), BpmnTaskAuditDTO.class);

3
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/AsyncTransferUserTaskJobHandler.java
View File

@ -12,7 +12,7 @@ import org.flowable.job.service.impl.persistence.entity.JobEntity;
import org.flowable.variable.api.delegate.VariableScope;
@Slf4j
public class AsyncTransferUserTaskJobHandler implements JobHandler {
public class AsyncTransferUserTaskJobHandler extends AbstractJobHandler implements JobHandler {
public static final String TYPE = "async-transfer-task";
@ -24,6 +24,7 @@ public class AsyncTransferUserTaskJobHandler implements JobHandler {
@Override
public void execute(JobEntity job, String configuration, VariableScope variableScope, CommandContext commandContext) {
log.info("AsyncTransferUserTaskJobHandler executing...");
log(job);
ProcessEngineConfigurationImpl processEngineConfiguration =
CommandContextUtil.getProcessEngineConfiguration(commandContext);
BpmnTaskTransferDTO dto = JSONUtil.toBean(job.getCustomValues(), BpmnTaskTransferDTO.class);

2
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/exception/handle/CustomAsyncRunnableExceptionExceptionHandler.java
View File

@ -32,7 +32,7 @@ public class CustomAsyncRunnableExceptionExceptionHandler implements AsyncRunnab
// Finally, Throw the exception to indicate the ExecuteAsyncJobCmd failed
String message = "Job " + job.getId() + " failed";
log.info(message, exception);
log.warn(message, exception);
if (job instanceof AbstractRuntimeJobEntity) {
AbstractRuntimeJobEntity runtimeJob = (AbstractRuntimeJobEntity) job;

11
workflow-engine-core/src/main/java/cn/axzo/workflow/core/engine/job/exception/handle/CustomWorkflowEngineExceptionHandler.java
View File

@ -18,9 +18,14 @@ import java.util.Objects;
public class CustomWorkflowEngineExceptionHandler implements AsyncRunnableExecutionExceptionHandler {
@Override
public boolean handleException(JobServiceConfiguration jobServiceConfiguration, JobInfo job, Throwable e) {
if (getRootCause(e).getClass().isAssignableFrom(WorkflowEngineException.class)) {
log.warn("AsyncApproveTaskJobHandler execute exception info: {}", e.getMessage(), e);
return true;
Throwable rootCause = getRootCause(e);
if (rootCause.getClass().isAssignableFrom(WorkflowEngineException.class)) {
WorkflowEngineException workflowEngineException = (WorkflowEngineException) rootCause;
if (Objects.equals(workflowEngineException.getCode(), "99806020")) {
log.info("AsyncApproveTaskJobHandler execute exception code: {} info: {}",
workflowEngineException.getCode(), rootCause.getMessage(), rootCause);
return true;
}
}
return false;
}

34
workflow-engine-server/src/main/java/cn/axzo/workflow/server/common/aspectj/ErrorReportAspect.java
View File

@ -43,8 +43,6 @@ public class ErrorReportAspect implements Ordered {
private String profile;
@Value("${workflow.sendDingTalk:true}")
private Boolean sendDingTalk;
@Value("${workflow.api.timeout:10}")
private Long apiTimeout;
@Resource
private ApplicationEventPublisher applicationEventPublisher;
@Resource
@ -66,22 +64,28 @@ public class ErrorReportAspect implements Ordered {
Signature signature = joinPoint.getSignature();
StopWatch watch = new StopWatch("running controller time");
watch.start(signature.toShortString());
Object result = joinPoint.proceed();
watch.stop();
if (!methodNames.contains(watch.getLastTaskName())) {
Object result = null;
try {
result = joinPoint.proceed();
} finally {
watch.stop();
log.info("StopWatch '{}': running time = {} 's", watch.getLastTaskName(), watch.getTotalTimeSeconds());
}
if (!signature.toShortString().contains("ExtAxApiLogServiceImpl")) {
String type = getType(joinPoint);
ApiLogEvent event = new ApiLogEvent(MDC.get(CTX_LOG_ID_MDC),
signature.toShortString(),
Objects.equals(type, "Controller") ? joinPoint.getArgs() : null,
Objects.equals(type, "Controller") ? result : null,
watch.getTotalTimeSeconds(),
type);
if (!methodNames.contains(watch.getLastTaskName())) {
log.info("StopWatch '{}': running time = {} 's", watch.getLastTaskName(), watch.getTotalTimeSeconds());
}
applicationEventPublisher.publishEvent(event);
if (!signature.toShortString().contains("ExtAxApiLogServiceImpl")) {
String type = getType(joinPoint);
ApiLogEvent event = new ApiLogEvent(MDC.get(CTX_LOG_ID_MDC),
signature.toShortString(),
Objects.equals(type, "Controller") ? joinPoint.getArgs() : null,
Objects.equals(type, "Controller") ? result : null,
watch.getTotalTimeSeconds(),
type);
applicationEventPublisher.publishEvent(event);
}
}
return result;
}