feat(openapi): 添加mk回调接口
This commit is contained in:
黄斌峰
-27
@@ -14,9 +14,6 @@ public interface IApi4MK {
|
||||
String API_PREFIX = "/openApi/mk";
|
||||
String PROCESS_API_PREFIX = API_PREFIX + "/process";
|
||||
String PROCESS_FINISH_CALLBACK = PROCESS_API_PREFIX + "/finishCallback";
|
||||
String PROCESS_APPROVAL_CALLBACK = PROCESS_API_PREFIX + "/approvalCallback";
|
||||
String PROCESS_REJECT_CALLBACK = PROCESS_API_PREFIX + "/rejectCallback";
|
||||
String PROCESS_REVOKE_CALLBACK = PROCESS_API_PREFIX + "/revokeCallback";
|
||||
String PROCESS_COMMON_CALLBACK = PROCESS_API_PREFIX + "/commonCallback";
|
||||
|
||||
/**
|
||||
@@ -34,28 +31,4 @@ public interface IApi4MK {
|
||||
*/
|
||||
@PostMapping(PROCESS_FINISH_CALLBACK)
|
||||
FR<Boolean> processFinishCallback(@RequestBody Api4MKProcessApprovalDTO param);
|
||||
|
||||
/**
|
||||
* 流程审批同意回调接口
|
||||
* @param param
|
||||
* @return
|
||||
*/
|
||||
@PostMapping(PROCESS_APPROVAL_CALLBACK)
|
||||
FR<Boolean> processApprovalCallback(@RequestBody Api4MKProcessApprovalDTO param);
|
||||
|
||||
/**
|
||||
* 流程审批拒绝回调接口
|
||||
* @param param
|
||||
* @return
|
||||
*/
|
||||
@PostMapping(PROCESS_REJECT_CALLBACK)
|
||||
FR<Boolean> processRejectCallback(@RequestBody Api4MKProcessApprovalDTO param);
|
||||
|
||||
/**
|
||||
* 流程撤销回调接口
|
||||
* @param param
|
||||
* @return
|
||||
*/
|
||||
@PostMapping(PROCESS_REVOKE_CALLBACK)
|
||||
FR<Boolean> processRevokeCallback(@RequestBody Api4MKProcessApprovalDTO param);
|
||||
}
|
||||
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
package org.springblade.openapi.mk.constant;
|
||||
|
||||
/**
|
||||
* 流程当前处理人相关redis锁key常量类
|
||||
* @author bfhuange
|
||||
* @since 2026/4/9
|
||||
*/
|
||||
public class ProcessLockKeyConstant {
|
||||
|
||||
/**
|
||||
* 任务缓存key前缀
|
||||
*/
|
||||
public static final String TASK_KEY_PREFIX = "process:cur-handler:task:";
|
||||
/**
|
||||
* 等待队列key
|
||||
*/
|
||||
public static final String WAITING_KEY = "process:cur-handler:waiting";
|
||||
/**
|
||||
* 流程锁key前缀
|
||||
*/
|
||||
public static final String PROCESS_LOCK_KEY_PREFIX = "process:cur-handler:lock:";
|
||||
/**
|
||||
* 派工锁key
|
||||
*/
|
||||
public static final String DISPATCH_LOCK_KEY = "process:cur-handler:dispatch-lock";
|
||||
/**
|
||||
* worker租约map key
|
||||
*/
|
||||
public static final String WORKER_LEASE_KEY = "process:cur-handler:worker-leases";
|
||||
}
|
||||
-9
@@ -69,13 +69,4 @@ public class Api4MKProcessApprovalDTO implements Serializable {
|
||||
*/
|
||||
private String operatorLoginName;
|
||||
|
||||
//====================非mk回调参数，回调接口设置参数===================
|
||||
/**
|
||||
* 是否流程已完成，非mk回调参数，回调接口设置参数
|
||||
*/
|
||||
private boolean complete;
|
||||
/**
|
||||
* 审批状态，非mk回调参数，回调接口设置参数
|
||||
*/
|
||||
private String approveStatus;
|
||||
}
|
||||
|
||||
-34
@@ -1,34 +0,0 @@
|
||||
package org.springblade.openapi.mk.pojo.dto;
|
||||
|
||||
import jakarta.validation.constraints.NotBlank;
|
||||
import lombok.Data;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* mk审批结束回调参数
|
||||
* @author bfhuange
|
||||
* @date 2024/9/5
|
||||
*/
|
||||
@Data
|
||||
public class ApiMKProcessFinishDTO implements Serializable {
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
/**
|
||||
* 流程实例id
|
||||
*/
|
||||
private String processInstanceId;
|
||||
/**
|
||||
* 表单实例id
|
||||
*/
|
||||
private String formInstanceId;
|
||||
/**
|
||||
* 模板编码，template_拼接 ProcessTypeEnum 的值
|
||||
*/
|
||||
private String templateCode;
|
||||
/**
|
||||
* 流程状态
|
||||
*/
|
||||
private String processStatus;
|
||||
}
|
||||
-40
@@ -1,40 +0,0 @@
|
||||
package org.springblade.openapi.mk.pojo.dto;
|
||||
|
||||
import io.swagger.v3.oas.annotations.media.Schema;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
@Builder
|
||||
@Data
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class ProcessApprovalDTO implements Serializable {
|
||||
|
||||
/**
|
||||
* 流程实例id
|
||||
*/
|
||||
@Schema(description = "流程实例id")
|
||||
protected String flowInstId;
|
||||
|
||||
|
||||
/**
|
||||
* 表单实例id
|
||||
*/
|
||||
protected String formInstanceId;
|
||||
|
||||
/**
|
||||
* 审批状态
|
||||
*/
|
||||
protected String approveStatus;
|
||||
|
||||
/**
|
||||
* 下级审批人
|
||||
*/
|
||||
protected String nextApproveUser;
|
||||
|
||||
|
||||
}
|
||||
+2
-2
@@ -49,10 +49,10 @@ public interface IBusinessProcessClient {
|
||||
/**
|
||||
* 修改业务流程状态
|
||||
* @param param
|
||||
* @return
|
||||
* @return 审批状态
|
||||
*/
|
||||
@PostMapping(UPDATE_BUSINESS_PROCESS_STATUS)
|
||||
FR<Boolean> updateBusinessProcessStatus(@Validated @RequestBody BusinessProcessUpdateDTO param);
|
||||
FR<String> updateBusinessProcessStatus(@Validated @RequestBody BusinessProcessUpdateDTO param);
|
||||
|
||||
/**
|
||||
* 修改业务流程审批人
|
||||
|
||||
-1
@@ -2,7 +2,6 @@ package org.springblade.process.pojo.dto;
|
||||
|
||||
import io.swagger.v3.oas.annotations.media.Schema;
|
||||
import lombok.Data;
|
||||
import org.springblade.system.pojo.dto.AdditionOperationParameterDTO;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.io.Serializable;
|
||||
|
||||
@@ -42,6 +42,10 @@
|
||||
<groupId>org.springblade</groupId>
|
||||
<artifactId>blade-mk-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springblade</groupId>
|
||||
<artifactId>blade-process-api</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.mapstruct</groupId>
|
||||
|
||||
+1
@@ -43,6 +43,7 @@ import org.springframework.context.annotation.ComponentScan;
|
||||
public class OpenApiApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
BladeApplication.disableNacosLaunchConfig();
|
||||
BladeApplication.run(AppConstant.APPLICATION_OPENAPI_NAME, OpenApiApplication.class, args);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
package org.springblade.openapi.mk;
|
||||
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import io.swagger.v3.oas.annotations.Hidden;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.springblade.core.log.exception.ServiceException;
|
||||
import org.springblade.core.tool.api.FR;
|
||||
import org.springblade.openapi.mk.api.IApi4MK;
|
||||
import org.springblade.openapi.mk.pojo.dto.Api4MKProcessApprovalDTO;
|
||||
import org.springblade.openapi.mk.pojo.enums.ProcessOperationType;
|
||||
import org.springblade.openapi.mk.support.base.ProcessHandler;
|
||||
import org.springblade.openapi.mk.util.ProcessTypeUtils;
|
||||
import org.springblade.thirdparty.mk.config.MKProperties;
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* 提供给mk的api实现类
|
||||
* @author bfhuange
|
||||
* @date 2024/9/9
|
||||
*/
|
||||
@Slf4j
|
||||
@Hidden
|
||||
@RestController
|
||||
public class Api4MK implements IApi4MK {
|
||||
private final MKProperties mkProperties;
|
||||
private final Map<String, ProcessHandler> handlerMap;
|
||||
|
||||
public Api4MK(MKProperties mkProperties, ObjectProvider<List<ProcessHandler>> handlersProvider) {
|
||||
this.mkProperties = mkProperties;
|
||||
handlerMap = handlersProvider.getIfAvailable(Collections::emptyList).stream()
|
||||
.flatMap(handler -> handler.getProcessTypes().stream()
|
||||
.collect(Collectors.toMap(Function.identity(), type -> handler, (a, b) -> {
|
||||
throw new ServiceException("重复的流程类型处理器");
|
||||
}))
|
||||
.entrySet()
|
||||
.stream())
|
||||
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (a, b) -> {
|
||||
throw new ServiceException("重复的流程类型处理器");
|
||||
}));
|
||||
}
|
||||
|
||||
@Override
|
||||
public FR<Boolean> processCommonCallback(Api4MKProcessApprovalDTO param) {
|
||||
log.info("mk流程通用回调 操作名称：{} 参数：{}", ProcessOperationType.getOperationName(param.getOperation()), JSON.toJSONString(param));
|
||||
callback(param, ProcessHandler::approve);
|
||||
return FR.status(true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public FR<Boolean> processFinishCallback(Api4MKProcessApprovalDTO param) {
|
||||
log.info("mk流程结束回调 参数：{}", JSON.toJSONString(param));
|
||||
// 手动设置操作类型，兼容历史接口
|
||||
param.setOperation(ProcessOperationType.PROCESS_FINISH);
|
||||
callback(param, ProcessHandler::approve);
|
||||
return FR.status(true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取处理器
|
||||
* @param processType
|
||||
* @return
|
||||
*/
|
||||
private ProcessHandler getHandler(String processType) {
|
||||
if (StringUtils.isBlank(processType)) {
|
||||
return null;
|
||||
}
|
||||
return handlerMap.get(processType);
|
||||
}
|
||||
|
||||
/**
|
||||
* 回调处理
|
||||
* @param param
|
||||
* @param consumer
|
||||
*/
|
||||
private void callback(Api4MKProcessApprovalDTO param, BiConsumer<ProcessHandler, Api4MKProcessApprovalDTO> consumer) {
|
||||
String processType = ProcessTypeUtils.getProcessType(param.getTemplateCode(), mkProperties.getTemplateCodePrefix());
|
||||
ProcessHandler handler = getHandler(processType);
|
||||
if (handler != null) {
|
||||
consumer.accept(handler, param);
|
||||
return;
|
||||
}
|
||||
log.warn("未配置流程类型对应的处理器 流程类型：{}", processType);
|
||||
}
|
||||
}
|
||||
+27
@@ -0,0 +1,27 @@
|
||||
package org.springblade.openapi.mk.config;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 当前处理人刷新相关异步线程池配置。
|
||||
*
|
||||
* @author bfhuange
|
||||
* @since 2026/4/9
|
||||
*/
|
||||
@Data
|
||||
@Component
|
||||
@ConfigurationProperties(prefix = "async")
|
||||
public class AsyncExecutorProperties {
|
||||
|
||||
/**
|
||||
* 当前处理人刷新工作线程池名称
|
||||
*/
|
||||
private String workerExecutorName = "mkRefreshWorkerExecutor";
|
||||
|
||||
/**
|
||||
* 当前处理人刷新调度线程池名称
|
||||
*/
|
||||
private String schedulerExecutorName = "mkRefreshSchedulerExecutor";
|
||||
}
|
||||
+52
@@ -0,0 +1,52 @@
|
||||
package org.springblade.openapi.mk.config;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* @author bfhuange
|
||||
* @since 2026/4/9
|
||||
*/
|
||||
@Component
|
||||
@ConfigurationProperties(prefix = "process.current-handler-refresh")
|
||||
@Data
|
||||
public class CurrentHandlerRefreshProperties {
|
||||
/**
|
||||
* 服务启动后的首次派工延迟，单位毫秒
|
||||
*/
|
||||
private long startupDispatchDelayMillis = 3000L;
|
||||
|
||||
/**
|
||||
* 首次执行延迟，单位秒
|
||||
*/
|
||||
private long initialDelaySeconds = 1;
|
||||
/**
|
||||
* 轮询间隔，单位秒
|
||||
*/
|
||||
private long intervalSeconds = 1;
|
||||
/**
|
||||
* 最大重试次数
|
||||
*/
|
||||
private int maxAttempts = 30;
|
||||
/**
|
||||
* 最大worker数
|
||||
*/
|
||||
private int maxWorkers = 5;
|
||||
/**
|
||||
* worker租约秒数
|
||||
*/
|
||||
private long workerLeaseSeconds =15;
|
||||
/**
|
||||
* 锁等待秒数
|
||||
*/
|
||||
private long lockWaitSeconds = 1;
|
||||
/**
|
||||
* 完成任务TTL
|
||||
*/
|
||||
private long doneTtlMinutes = 5L;
|
||||
/**
|
||||
* 失败任务TTL
|
||||
*/
|
||||
private long failedTtlMinutes = 30L;
|
||||
}
|
||||
+301
@@ -0,0 +1,301 @@
|
||||
package org.springblade.openapi.mk.support.base;
|
||||
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springblade.core.log.exception.ServiceException;
|
||||
import org.springblade.core.tool.api.FR;
|
||||
import org.springblade.core.tool.utils.StringUtil;
|
||||
import org.springblade.openapi.mk.pojo.dto.Api4MKProcessApprovalDTO;
|
||||
import org.springblade.openapi.mk.pojo.enums.ProcessCallbackType;
|
||||
import org.springblade.openapi.mk.support.handler.ProcessCurrentHandlerRefreshService;
|
||||
import org.springblade.openapi.mk.util.ProcessTypeUtils;
|
||||
import org.springblade.process.feign.IBusinessProcessClient;
|
||||
import org.springblade.process.pojo.dto.BusinessProcessUpdateDTO;
|
||||
import org.springblade.process.pojo.enums.ApproveStatusEnum;
|
||||
import org.springblade.process.pojo.vo.BusinessProcessVO;
|
||||
import org.springblade.thirdparty.mk.config.MKProperties;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
/**
|
||||
* 抽象流程操作处理器，实现公共逻辑
|
||||
* @author bfhuange
|
||||
* @date 2024/9/9
|
||||
*/
|
||||
@Slf4j
|
||||
public abstract class AbstractProcessOperationHandler implements ProcessHandler, ProcessOperationHandler {
|
||||
|
||||
@Autowired
|
||||
protected IBusinessProcessClient processClient;
|
||||
|
||||
@Autowired
|
||||
protected ProcessCurrentHandlerRefreshService refreshService;
|
||||
|
||||
@Autowired
|
||||
protected MKProperties mkProperties;
|
||||
|
||||
@Override
|
||||
public List<String> getProcessTypes() {
|
||||
return List.of(this.getProcessType());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void approve(Api4MKProcessApprovalDTO param) {
|
||||
// 入口层只接收 MK 原始回调参数，随后统一组装为内部上下文对象，
|
||||
// 把流程类型、审批状态、是否完成、是否异步等内部处理语义集中收口在这里。
|
||||
ProcessCallbackType callbackType = ProcessCallbackType.getCallbackType(param.getOperation());
|
||||
if (callbackType == null) {
|
||||
log.error("未配置事件的操作：{}", param.getOperation());
|
||||
return;
|
||||
}
|
||||
switch (callbackType) {
|
||||
// 提交
|
||||
case SUBMIT -> submit(buildSubmitContext(param));
|
||||
// 审批结束
|
||||
case FINISH -> approveFinish(buildFinishContext(param));
|
||||
// 撤回
|
||||
case RETRACT -> approveRevoke(buildRevokeContext(param));
|
||||
// 通过
|
||||
case PASS -> approvePass(buildPassContext(param));
|
||||
// 驳回
|
||||
case REJECT -> approveReject(buildRejectContext(param));
|
||||
// 废弃
|
||||
case ABANDON -> approveAbandon(buildAbandonContext(param));
|
||||
// 修改当前处理人
|
||||
case CHANGE_CUR_HANDLER -> handleCommon(buildChangeCurrentHandlerContext(param));
|
||||
default -> log.error("未配置事件的操作：{}", param.getOperation());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void approveFinish(ProcessOperationContext param) {
|
||||
approveCommon(param, this::approveFinishBusiness);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void approvePass(ProcessOperationContext param) {
|
||||
approveCommon(param, this::approvePassBusiness);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void approveReject(ProcessOperationContext param) {
|
||||
approveCommon(param, this::approveRejectBusiness);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void approveRevoke(ProcessOperationContext param) {
|
||||
approveCommon(param, this::approveRevokeBusiness);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理提交
|
||||
* @param param
|
||||
*/
|
||||
@Override
|
||||
public void submit(ProcessOperationContext param) {
|
||||
// 一般提交后只需要更新当前处理人
|
||||
handleCommon(param);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理废弃
|
||||
* @param param
|
||||
*/
|
||||
@Override
|
||||
public void approveAbandon(ProcessOperationContext param) {
|
||||
approveCommon(param, this::approveAbandonBusiness);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理审批通用逻辑
|
||||
* @param param
|
||||
* @param businessHandler
|
||||
*/
|
||||
protected void approveCommon(ProcessOperationContext param, Consumer<ProcessOperationContext> businessHandler) {
|
||||
// 1. 更新流程状态
|
||||
updateBusinessProcessStatus(param);
|
||||
// 2. 同步处理业务逻辑
|
||||
businessHandler.accept(param);
|
||||
// 3. 处理当前处理人刷新
|
||||
handleCommon(param);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理公共异步逻辑
|
||||
* @param param
|
||||
*/
|
||||
protected void handleCommon(ProcessOperationContext param) {
|
||||
// 当前处理人支持按事件选择同步刷新或任务调度刷新
|
||||
if (param.isAsync()) {
|
||||
refreshService.enqueue(param);
|
||||
} else {
|
||||
refreshService.refreshNow(param);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新流程状态
|
||||
* @param param
|
||||
*/
|
||||
private void updateBusinessProcessStatus(ProcessOperationContext param) {
|
||||
// 更新流程状态
|
||||
BusinessProcessUpdateDTO updateStatusParam = getBusinessProcessUpdateParam(param);
|
||||
FR<String> statusResult = processClient.updateBusinessProcessStatus(updateStatusParam);
|
||||
if (FR.isNotSuccess(statusResult)) {
|
||||
log.error("更新流程状态异常 ：{}", JSON.toJSONString(statusResult));
|
||||
String errorMessage = Optional.ofNullable(statusResult)
|
||||
.map(FR::getMsg)
|
||||
.orElse("");
|
||||
throw new ServiceException("更新流程状态异常：" + errorMessage);
|
||||
}
|
||||
// 具体审批状态要以更新 BusinessProcess 返回的为准，有些比如驳回到上一个审批节点（非起草节点）的，不需要更新状态
|
||||
String approveStatus = statusResult.getData();
|
||||
if (StringUtil.isBlank(approveStatus)) {
|
||||
param.setApproveStatus(null);
|
||||
} else {
|
||||
param.setApproveStatus(approveStatus);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取流程更新参数
|
||||
* @param param
|
||||
* @return
|
||||
*/
|
||||
private BusinessProcessUpdateDTO getBusinessProcessUpdateParam(ProcessOperationContext param) {
|
||||
BusinessProcessUpdateDTO updateParam = new BusinessProcessUpdateDTO();
|
||||
updateParam.setProcessInstanceId(param.getProcessInstanceId());
|
||||
updateParam.setPromoterLoginName(param.getApplicantLoginName());
|
||||
updateParam.setOperationNodeId(param.getCurrentNodeId());
|
||||
updateParam.setOperationNodeNumber(param.getCurrentNodeNumber());
|
||||
updateParam.setComplete(param.isComplete());
|
||||
updateParam.setApproveStatus(param.getApproveStatus());
|
||||
updateParam.setRejectNodeId(param.getRejectNodeId());
|
||||
return updateParam;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造提交流程上下文。
|
||||
*/
|
||||
protected ProcessOperationContext buildSubmitContext(Api4MKProcessApprovalDTO callbackParam) {
|
||||
return buildContext(callbackParam, ApproveStatusEnum.APPROVING.getValue(), false, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造审批通过上下文。
|
||||
*/
|
||||
protected ProcessOperationContext buildPassContext(Api4MKProcessApprovalDTO callbackParam) {
|
||||
return buildContext(callbackParam, ApproveStatusEnum.APPROVING.getValue(), false, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造流程结束上下文。
|
||||
*/
|
||||
protected ProcessOperationContext buildFinishContext(Api4MKProcessApprovalDTO callbackParam) {
|
||||
return buildContext(callbackParam, ApproveStatusEnum.APPROVED.getValue(), true, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造驳回上下文。
|
||||
*/
|
||||
protected ProcessOperationContext buildRejectContext(Api4MKProcessApprovalDTO callbackParam) {
|
||||
return buildContext(callbackParam, ApproveStatusEnum.REJECTED.getValue(), false, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造撤回上下文。
|
||||
*/
|
||||
protected ProcessOperationContext buildRevokeContext(Api4MKProcessApprovalDTO callbackParam) {
|
||||
return buildContext(callbackParam, ApproveStatusEnum.REVOCATION.getValue(), false, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造废弃上下文。
|
||||
*/
|
||||
protected ProcessOperationContext buildAbandonContext(Api4MKProcessApprovalDTO callbackParam) {
|
||||
return buildContext(callbackParam, ApproveStatusEnum.ABANDON.getValue(), false, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造仅刷新当前处理人的上下文。
|
||||
*/
|
||||
protected ProcessOperationContext buildChangeCurrentHandlerContext(Api4MKProcessApprovalDTO callbackParam) {
|
||||
return buildContext(callbackParam, null, false, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造流程内部处理上下文。
|
||||
* 这里统一固化 processType，避免后续业务处理和异步刷新阶段再次根据模板编码反推。
|
||||
*/
|
||||
protected ProcessOperationContext buildContext(Api4MKProcessApprovalDTO callbackParam,
|
||||
String approveStatus,
|
||||
boolean complete,
|
||||
boolean async) {
|
||||
return ProcessOperationContext.builder()
|
||||
.callbackParam(callbackParam)
|
||||
.processType(ProcessTypeUtils.getProcessType(callbackParam.getTemplateCode(), mkProperties.getTemplateCodePrefix()))
|
||||
.approveStatus(approveStatus)
|
||||
.complete(complete)
|
||||
.async(async)
|
||||
.build();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取流程类型
|
||||
* @return
|
||||
*/
|
||||
protected String getProcessType() {
|
||||
throw new ServiceException("未配置流程类型");
|
||||
};
|
||||
|
||||
/**
|
||||
* 当前处理人刷新成功后回调各业务模块
|
||||
* @param param 回调参数
|
||||
* @param businessProcessVO 最新流程快照
|
||||
*/
|
||||
public void handleCurrentHandlerRefresh(ProcessOperationContext param, BusinessProcessVO businessProcessVO) {
|
||||
if (businessProcessVO != null) {
|
||||
this.commonBusiness(param, businessProcessVO);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理公共业务逻辑
|
||||
* @param param
|
||||
* @param businessProcessVO
|
||||
*/
|
||||
protected abstract void commonBusiness(ProcessOperationContext param, BusinessProcessVO businessProcessVO);
|
||||
|
||||
/**
|
||||
* 处理审批通过同步逻辑
|
||||
* @param param
|
||||
*/
|
||||
protected abstract void approvePassBusiness(ProcessOperationContext param);
|
||||
|
||||
/**
|
||||
* 处理流程结束同步逻辑
|
||||
* @param param
|
||||
*/
|
||||
protected abstract void approveFinishBusiness(ProcessOperationContext param);
|
||||
|
||||
/**
|
||||
* 处理审批驳回同步逻辑
|
||||
* @param param
|
||||
*/
|
||||
protected abstract void approveRejectBusiness(ProcessOperationContext param);
|
||||
|
||||
/**
|
||||
* 处理审批撤回同步逻辑
|
||||
* @param param
|
||||
*/
|
||||
protected abstract void approveRevokeBusiness(ProcessOperationContext param);
|
||||
|
||||
/**
|
||||
* 处理审批废弃同步逻辑 todo 为了避免代码报错，先用空实现
|
||||
* @param param
|
||||
*/
|
||||
protected void approveAbandonBusiness(ProcessOperationContext param) {};
|
||||
}
|
||||
+27
@@ -0,0 +1,27 @@
|
||||
package org.springblade.openapi.mk.support.base;
|
||||
|
||||
|
||||
import org.springblade.openapi.mk.pojo.dto.Api4MKProcessApprovalDTO;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 流程处理器
|
||||
* @author bfhuange
|
||||
* @date 2024/9/9
|
||||
*/
|
||||
public interface ProcessHandler {
|
||||
|
||||
/**
|
||||
* 获取流程类型列表
|
||||
* @return
|
||||
*/
|
||||
List<String> getProcessTypes();
|
||||
|
||||
/**
|
||||
* 通用审批
|
||||
* @param param
|
||||
*/
|
||||
void approve(Api4MKProcessApprovalDTO param);
|
||||
|
||||
}
|
||||
+139
@@ -0,0 +1,139 @@
|
||||
package org.springblade.openapi.mk.support.base;
|
||||
|
||||
import com.alibaba.fastjson2.annotation.JSONField;
|
||||
import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import org.springblade.openapi.mk.pojo.dto.Api4MKProcessApprovalDTO;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* 流程回调内部处理上下文。
|
||||
* <p>
|
||||
* callbackParam 仅保留 MK 原始回调参数，
|
||||
* 其余字段为 openapi 在处理过程中补充的上下文参数。
|
||||
* </p>
|
||||
* <p>
|
||||
* 设计目的：
|
||||
* 1. 避免把内部推导字段继续堆到 MK 原始回调 DTO 上；
|
||||
* 2. 对外保留原始回调对象，便于排查问题、记录日志和后续扩展；
|
||||
* 3. 通过代理 getter 尽量兼容原来直接读取 DTO 字段的使用习惯，降低老流程和后续分支合并成本。
|
||||
* </p>
|
||||
*
|
||||
* @author bfhuange
|
||||
* @date 2026/4/9
|
||||
*/
|
||||
@Data
|
||||
@Builder
|
||||
public class ProcessOperationContext implements Serializable {
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
/**
|
||||
* MK 原始回调参数
|
||||
*/
|
||||
private Api4MKProcessApprovalDTO callbackParam;
|
||||
|
||||
/**
|
||||
* 流程类型
|
||||
*/
|
||||
private String processType;
|
||||
|
||||
/**
|
||||
* 业务审批状态。
|
||||
* 这是系统内部按事件语义统一补充的状态，不属于 MK 原始回调参数。
|
||||
*/
|
||||
private String approveStatus;
|
||||
|
||||
/**
|
||||
* 是否流程已完成。
|
||||
* 这是系统内部按事件语义统一补充的状态，不属于 MK 原始回调参数。
|
||||
*/
|
||||
private boolean complete;
|
||||
|
||||
/**
|
||||
* 是否异步刷新当前处理人。
|
||||
* 用于控制当前处理人更新是走同步刷新还是异步调度任务。
|
||||
*/
|
||||
private boolean async;
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getProcessInstanceId() {
|
||||
return callbackParam == null ? null : callbackParam.getProcessInstanceId();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getFormInstanceId() {
|
||||
return callbackParam == null ? null : callbackParam.getFormInstanceId();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getTemplateId() {
|
||||
return callbackParam == null ? null : callbackParam.getTemplateId();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getTemplateCode() {
|
||||
return callbackParam == null ? null : callbackParam.getTemplateCode();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getProcessStatus() {
|
||||
return callbackParam == null ? null : callbackParam.getProcessStatus();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getApplicantLoginName() {
|
||||
return callbackParam == null ? null : callbackParam.getApplicantLoginName();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getRejectNodeId() {
|
||||
return callbackParam == null ? null : callbackParam.getRejectNodeId();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getCurrentNodeId() {
|
||||
return callbackParam == null ? null : callbackParam.getCurrentNodeId();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getCurrentNodeNumber() {
|
||||
return callbackParam == null ? null : callbackParam.getCurrentNodeNumber();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getOperation() {
|
||||
return callbackParam == null ? null : callbackParam.getOperation();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getOperationName() {
|
||||
return callbackParam == null ? null : callbackParam.getOperationName();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getApprovalOpinion() {
|
||||
return callbackParam == null ? null : callbackParam.getApprovalOpinion();
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
@JSONField(serialize = false)
|
||||
public String getOperatorLoginName() {
|
||||
return callbackParam == null ? null : callbackParam.getOperatorLoginName();
|
||||
}
|
||||
}
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
package org.springblade.openapi.mk.support.base;
|
||||
|
||||
/**
|
||||
* 流程操作处理器
|
||||
* @author bfhuange
|
||||
* @since 2024/11/25
|
||||
*/
|
||||
public interface ProcessOperationHandler {
|
||||
|
||||
/**
|
||||
* 提交
|
||||
* @param param
|
||||
*/
|
||||
void submit(ProcessOperationContext param);
|
||||
|
||||
/**
|
||||
* 审批结束
|
||||
* @param param
|
||||
*/
|
||||
void approveFinish(ProcessOperationContext param);
|
||||
|
||||
/**
|
||||
* 审批同意
|
||||
* @param param
|
||||
*/
|
||||
void approvePass(ProcessOperationContext param);
|
||||
|
||||
/**
|
||||
* 审批拒绝
|
||||
* @param param
|
||||
*/
|
||||
void approveReject(ProcessOperationContext param);
|
||||
|
||||
/**
|
||||
* 审批撤销
|
||||
* @param param
|
||||
*/
|
||||
void approveRevoke(ProcessOperationContext param);
|
||||
|
||||
/**
|
||||
* 审批废弃
|
||||
* @param param
|
||||
*/
|
||||
void approveAbandon(ProcessOperationContext param);
|
||||
}
|
||||
+113
@@ -0,0 +1,113 @@
|
||||
package org.springblade.openapi.mk.support.handler;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.dromara.dynamictp.core.DtpRegistry;
|
||||
import org.dromara.dynamictp.core.aware.TaskEnhanceAware;
|
||||
import org.springblade.core.log.exception.ServiceException;
|
||||
import org.springblade.openapi.mk.config.AsyncExecutorProperties;
|
||||
import org.springframework.boot.context.event.ApplicationReadyEvent;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.core.Ordered;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* @author bfhuange
|
||||
* @date 2024/9/20
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class AsyncService {
|
||||
private final AsyncExecutorProperties asyncExecutorProperties;
|
||||
private Executor workerExecutor;
|
||||
private ScheduledExecutorService schedulerExecutor;
|
||||
|
||||
public AsyncService(AsyncExecutorProperties asyncExecutorProperties) {
|
||||
this.asyncExecutorProperties = asyncExecutorProperties;
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动完成后预加载并校验线程池配置，避免等到第一次真正执行任务时才发现线程池缺失或类型配置错误。
|
||||
*/
|
||||
@Order(Ordered.HIGHEST_PRECEDENCE)
|
||||
@EventListener(ApplicationReadyEvent.class)
|
||||
public void initExecutors() {
|
||||
this.workerExecutor = resolveWorkerExecutor();
|
||||
this.schedulerExecutor = resolveSchedulerExecutor();
|
||||
log.info("当前处理人刷新异步线程池初始化完成，workerExecutorName：{}，schedulerExecutorName：{}",
|
||||
asyncExecutorProperties.getWorkerExecutorName(), asyncExecutorProperties.getSchedulerExecutorName());
|
||||
}
|
||||
|
||||
/**
|
||||
* 立即异步执行
|
||||
* @param runnable 任务
|
||||
*/
|
||||
public void execute(Runnable runnable) {
|
||||
getWorkerExecutor().execute(runnable);
|
||||
}
|
||||
|
||||
/**
|
||||
* 延迟执行指定毫秒数。
|
||||
* <p>
|
||||
* 这里改为使用 ScheduledDtpExecutor 做真正的定时调度，
|
||||
* 避免再通过线程池线程 sleep 的方式占用工作线程，导致真正的业务任务迟迟无法启动。
|
||||
* </p>
|
||||
*
|
||||
* @param delayMillis 延迟毫秒数
|
||||
* @param runnable 任务
|
||||
*/
|
||||
public void delayExecute(long delayMillis, Runnable runnable) {
|
||||
if (delayMillis <= 0) {
|
||||
execute(runnable);
|
||||
return;
|
||||
}
|
||||
Runnable dispatchRunnable = wrapWithConfiguredTaskWrappers(
|
||||
asyncExecutorProperties.getSchedulerExecutorName(),
|
||||
() -> execute(runnable)
|
||||
);
|
||||
getSchedulerExecutor().schedule(dispatchRunnable, delayMillis, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
private Executor getWorkerExecutor() {
|
||||
return workerExecutor != null ? workerExecutor : resolveWorkerExecutor();
|
||||
}
|
||||
|
||||
private ScheduledExecutorService getSchedulerExecutor() {
|
||||
return schedulerExecutor != null ? schedulerExecutor : resolveSchedulerExecutor();
|
||||
}
|
||||
|
||||
private Executor resolveWorkerExecutor() {
|
||||
return DtpRegistry.getExecutor(asyncExecutorProperties.getWorkerExecutorName());
|
||||
}
|
||||
|
||||
private ScheduledExecutorService resolveSchedulerExecutor() {
|
||||
String schedulerExecutorName = asyncExecutorProperties.getSchedulerExecutorName();
|
||||
Executor executor = DtpRegistry.getExecutor(schedulerExecutorName);
|
||||
if (executor instanceof ScheduledExecutorService scheduledExecutorService) {
|
||||
return scheduledExecutorService;
|
||||
}
|
||||
String message = "线程池未按 ScheduledExecutorService 注册，name: " + schedulerExecutorName;
|
||||
log.error(message);
|
||||
throw new ServiceException(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 按线程池已配置的 task wrappers 手动包装任务。
|
||||
* <p>
|
||||
* 当前使用的 dynamic-tp 版本下，ScheduledDtpExecutor 对 taskWrapper 的透传存在缺口，
|
||||
* 这里直接读取线程池上已生效的 wrappers，按框架默认增强链顺序主动包装一次，
|
||||
* 这样既能复用现有配置，又避免手写 mdc 透传逻辑与框架实现产生偏差。
|
||||
* </p>
|
||||
*/
|
||||
private Runnable wrapWithConfiguredTaskWrappers(String executorName, Runnable runnable) {
|
||||
Executor executor = DtpRegistry.getExecutor(executorName);
|
||||
if (executor instanceof TaskEnhanceAware taskEnhanceAware) {
|
||||
return taskEnhanceAware.getEnhancedTask(runnable, taskEnhanceAware.getTaskWrappers());
|
||||
}
|
||||
return runnable;
|
||||
}
|
||||
}
|
||||
+903
@@ -0,0 +1,903 @@
|
||||
package org.springblade.openapi.mk.support.handler;
|
||||
|
||||
import cn.hutool.core.util.IdUtil;
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.redisson.api.RLock;
|
||||
import org.redisson.api.RMapCache;
|
||||
import org.springblade.core.redis.cache.BladeRedis;
|
||||
import org.springblade.core.redis.lock.RedisLockClient;
|
||||
import org.springblade.core.tool.api.FR;
|
||||
import org.springblade.openapi.mk.config.CurrentHandlerRefreshProperties;
|
||||
import org.springblade.openapi.mk.constant.ProcessLockKeyConstant;
|
||||
import org.springblade.openapi.mk.pojo.enums.ProcessCallbackType;
|
||||
import org.springblade.openapi.mk.support.base.AbstractProcessOperationHandler;
|
||||
import org.springblade.openapi.mk.support.base.ProcessOperationContext;
|
||||
import org.springblade.process.feign.IBusinessProcessClient;
|
||||
import org.springblade.process.pojo.dto.BusinessProcessCurrentHandlerRefreshDTO;
|
||||
import org.springblade.process.pojo.vo.BusinessProcessVO;
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.boot.context.event.ApplicationReadyEvent;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
/**
|
||||
* 当前处理人刷新调度服务。
|
||||
* <p>
|
||||
* 背景：
|
||||
* 流程引擎回调业务系统时，流程往往还没有真正流转到下一个激活节点，
|
||||
* 此时立即查询当前节点/当前处理人，拿到的仍可能是上一节点的旧结果。
|
||||
* 因此这里不再依赖一次性的固定延迟，而是改成“按流程实例维度入队 + 固定间隔轮询刷新”的调度模型。
|
||||
* </p>
|
||||
* <p>
|
||||
* 整体流程：
|
||||
* 1. openapi 收到流程事件后，先同步更新业务流程状态；
|
||||
* 2. 如果当前事件要求异步刷新当前处理人，则调用 {@link #enqueue(ProcessOperationContext)} 写入刷新任务；
|
||||
* 3. 任务以流程实例 id 为唯一主记录保存在 Redis，记录期望版本、执行版本、基线快照、最近回调参数、重试次数等信息；
|
||||
* 4. 同一个流程实例只保留一条主任务记录，新的回调不会重复创建任务，只会提升 {@code desiredVersion} 并覆盖最近一次回调参数；
|
||||
* 5. 等待中的流程实例 id 会放入 Redis ZSet，score 为下次重试时间，用于按时间顺序派工；
|
||||
* 6. 调度器 {@link #tryDispatch()} 会在集群范围内抢占派工锁，按配置的最大 worker 数拉起异步 worker；
|
||||
* 7. worker 执行时调用 system 侧“只刷新当前节点/当前处理人”接口，并比较“当前节点 + 当前处理人”快照是否相对基线发生变化；
|
||||
* 8. 如果快照未变化，说明流程大概率还没流转完成，则按固定间隔重新入队重试；
|
||||
* 9. 如果快照发生变化，则回调对应业务处理器 {@link AbstractProcessOperationHandler#handleCurrentHandlerRefresh(ProcessOperationContext, BusinessProcessVO)}；
|
||||
* 10. 若执行期间又收到同一流程的新回调，则旧版本执行完后会把最新快照提升为新基线，并重新排到队尾，避免同一流程长期占用 worker；
|
||||
* 11. 当达到最大重试次数后，任务进入失败态并保留一段时间，便于排查；
|
||||
* 12. 成功完成的任务进入完成态并短期保留，随后自动过期。
|
||||
* </p>
|
||||
* <p>
|
||||
* 集群与并发约束：
|
||||
* 1. 流程实例级别使用分布式锁，保证同一流程实例的任务状态变更串行化；
|
||||
* 2. 派工使用全局分布式锁，保证多个实例不会同时超发 worker；
|
||||
* 3. 活跃 worker 数通过 Redis 租约控制，服务异常中断后，租约超时即可视为 worker 失活；
|
||||
* 4. 运行中的任务会持续更新心跳，如果服务升级、中断或线程异常退出，超时恢复逻辑会把任务重新转回等待态；
|
||||
* 5. 启动时不会全量恢复运行中任务，避免在集群环境中误伤其他实例上仍在执行的任务。
|
||||
* </p>
|
||||
* <p>
|
||||
* 成功判定规则：
|
||||
* 不再区分终态/非终态，也不依赖回调里传入的 complete true/false 单独判定是否成功，
|
||||
* 统一以“当前节点变化 + 当前处理人变化后的最新快照”是否相对基线发生变化作为刷新成功依据。
|
||||
* </p>
|
||||
*
|
||||
* @author bfhuange
|
||||
* @date 2026/4/9
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class ProcessCurrentHandlerRefreshService {
|
||||
|
||||
|
||||
private final AsyncService asyncService;
|
||||
private final BladeRedis bladeRedis;
|
||||
private final RedisLockClient redisLockClient;
|
||||
private final IBusinessProcessClient processClient;
|
||||
private final CurrentHandlerRefreshProperties refreshProperties;
|
||||
private final ObjectProvider<List<AbstractProcessOperationHandler>> handlersProvider;
|
||||
private final Map<String, AbstractProcessOperationHandler> handlerMap;
|
||||
|
||||
public ProcessCurrentHandlerRefreshService(AsyncService asyncService,
|
||||
BladeRedis bladeRedis,
|
||||
RedisLockClient redisLockClient,
|
||||
IBusinessProcessClient processClient,
|
||||
CurrentHandlerRefreshProperties refreshProperties,
|
||||
ObjectProvider<List<AbstractProcessOperationHandler>> handlersProvider) {
|
||||
this.asyncService = asyncService;
|
||||
this.bladeRedis = bladeRedis;
|
||||
this.redisLockClient = redisLockClient;
|
||||
this.processClient = processClient;
|
||||
this.refreshProperties = refreshProperties;
|
||||
this.handlersProvider = handlersProvider;
|
||||
this.handlerMap = new ConcurrentHashMap<>();
|
||||
}
|
||||
|
||||
/**
|
||||
* 服务启动后恢复未完成任务
|
||||
*/
|
||||
@Order
|
||||
@EventListener(ApplicationReadyEvent.class)
|
||||
public void init() {
|
||||
// 集群环境下不能在启动时无差别回收所有运行中任务，否则会误伤其他实例正在执行的任务
|
||||
asyncService.delayExecute(refreshProperties.getStartupDispatchDelayMillis(), this::tryDispatch);
|
||||
}
|
||||
|
||||
/**
|
||||
* 按流程类型懒加载处理器，避免在bean初始化阶段提前拉起handler导致循环依赖
|
||||
*/
|
||||
private AbstractProcessOperationHandler getHandler(String processType) {
|
||||
if (StringUtils.isBlank(processType)) {
|
||||
return null;
|
||||
}
|
||||
if (handlerMap.isEmpty()) {
|
||||
synchronized (this) {
|
||||
if (handlerMap.isEmpty()) {
|
||||
List<AbstractProcessOperationHandler> handlers = handlersProvider.getIfAvailable(Collections::emptyList);
|
||||
handlers.forEach(handler -> handler.getProcessTypes()
|
||||
.forEach(type -> this.handlerMap.put(type, handler)));
|
||||
}
|
||||
}
|
||||
}
|
||||
return handlerMap.get(processType);
|
||||
}
|
||||
|
||||
/**
|
||||
* 写入刷新任务
|
||||
*
|
||||
* @param param 回调参数
|
||||
*/
|
||||
public void enqueue(ProcessOperationContext param) {
|
||||
if (param == null || StringUtils.isAnyBlank(param.getProcessType(), param.getProcessInstanceId())) {
|
||||
log.warn("当前处理人刷新任务入队失败，上下文为空或流程类型/流程实例id为空，param：{}", JSON.toJSONString(param));
|
||||
return;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
String processInstanceId = param.getProcessInstanceId();
|
||||
RLock lock = getProcessLock(processInstanceId);
|
||||
boolean locked = false;
|
||||
try {
|
||||
locked = lock.tryLock(refreshProperties.getLockWaitSeconds(), TimeUnit.SECONDS);
|
||||
if (!locked) {
|
||||
log.warn("获取流程刷新任务锁失败，流程实例id：{}", processInstanceId);
|
||||
return;
|
||||
}
|
||||
ProcessCurrentHandlerRefreshTask task = getTask(processInstanceId);
|
||||
if (task == null) {
|
||||
task = new ProcessCurrentHandlerRefreshTask();
|
||||
task.setProcessInstanceId(processInstanceId);
|
||||
task.setState(TaskState.STATE_WAITING);
|
||||
}
|
||||
task.setProcessType(param.getProcessType());
|
||||
task.setContext(param);
|
||||
task.setDesiredVersion(task.getDesiredVersion() + 1);
|
||||
task.setLastCallbackAt(now);
|
||||
if (!TaskState.STATE_RUNNING.equals(task.getState())) {
|
||||
// 非运行中任务表示上一轮刷新周期已经结束或尚未开始。
|
||||
// 这里必须按本次回调重新建立基线，避免撤回后再次提交时沿用上一轮旧快照，导致新一轮刷新永远无法命中成功条件。
|
||||
task.setBaselineSnapshot(queryCurrentSnapshot(processInstanceId));
|
||||
task.setLatestSnapshot(null);
|
||||
task.setLastSuccessAt(null);
|
||||
task.setState(TaskState.STATE_WAITING);
|
||||
task.setAttemptCount(0);
|
||||
task.setProcessingVersion(0);
|
||||
task.setRunToken(null);
|
||||
task.setStartedAt(null);
|
||||
task.setHeartbeatAt(null);
|
||||
task.setNextRetryAt(now + initialDelayMillis());
|
||||
saveTask(task);
|
||||
putWaitingTask(processInstanceId, task.getNextRetryAt());
|
||||
log.info("当前处理人刷新任务入队，流程实例id：{}，{}", processInstanceId, formatTaskLog(task));
|
||||
scheduleDispatch(initialDelayMillis());
|
||||
} else {
|
||||
saveTask(task);
|
||||
log.info("当前处理人刷新任务更新执行中版本，流程实例id：{}，{}", processInstanceId, formatTaskLog(task));
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("当前处理人刷新任务入队被中断，流程实例id：{}", processInstanceId, e);
|
||||
} catch (Exception e) {
|
||||
log.error("当前处理人刷新任务入队异常，流程实例id：{}", processInstanceId, e);
|
||||
} finally {
|
||||
unlock(lock);
|
||||
}
|
||||
tryDispatch();
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步立即刷新当前处理人
|
||||
*
|
||||
* @param param 回调参数
|
||||
*/
|
||||
public void refreshNow(ProcessOperationContext param) {
|
||||
if (param == null || StringUtils.isAnyBlank(param.getProcessType(), param.getProcessInstanceId())) {
|
||||
log.warn("同步刷新当前处理人失败，上下文为空或流程类型/流程实例id为空，param：{}", JSON.toJSONString(param));
|
||||
return;
|
||||
}
|
||||
FR<BusinessProcessVO> result = processClient.refreshBusinessProcessCurrentHandlers(buildUpdateParam(param));
|
||||
if (result == null || FR.isNotSuccess(result)) {
|
||||
log.warn("同步刷新当前处理人失败，转入异步队列重试，流程实例id：{} result：{}", param.getProcessInstanceId(), JSON.toJSONString(result));
|
||||
enqueue(param);
|
||||
return;
|
||||
}
|
||||
AbstractProcessOperationHandler handler = getHandler(param.getProcessType());
|
||||
if (handler == null) {
|
||||
log.error("同步刷新当前处理人失败，未找到处理器，流程类型：{} 流程实例id：{}", param.getProcessType(), param.getProcessInstanceId());
|
||||
return;
|
||||
}
|
||||
handler.handleCurrentHandlerRefresh(param, result.getData());
|
||||
}
|
||||
|
||||
/**
|
||||
* 派发worker执行任务
|
||||
*/
|
||||
public void tryDispatch() {
|
||||
int activeWorkerCount = getActiveWorkerCount();
|
||||
if (activeWorkerCount >= refreshProperties.getMaxWorkers()) {
|
||||
log.info("当前处理人刷新派工跳过，活跃worker已满，activeWorkers：{}，maxWorkers：{}", activeWorkerCount, refreshProperties.getMaxWorkers());
|
||||
return;
|
||||
}
|
||||
RLock dispatchLock = getDispatchLock();
|
||||
boolean locked = false;
|
||||
try {
|
||||
locked = dispatchLock.tryLock(refreshProperties.getLockWaitSeconds(), TimeUnit.SECONDS);
|
||||
if (!locked) {
|
||||
return;
|
||||
}
|
||||
// 先恢复真正超时的运行中任务，再判断是否有可执行任务
|
||||
recoverTimeoutTasks();
|
||||
if (!hasWaitingTask()) {
|
||||
return;
|
||||
}
|
||||
activeWorkerCount = getActiveWorkerCount();
|
||||
if (activeWorkerCount >= refreshProperties.getMaxWorkers()) {
|
||||
log.info("当前处理人刷新派工二次检查跳过，活跃worker已满，activeWorkers：{}，maxWorkers：{}", activeWorkerCount, refreshProperties.getMaxWorkers());
|
||||
return;
|
||||
}
|
||||
while ((activeWorkerCount = getActiveWorkerCount()) < refreshProperties.getMaxWorkers()) {
|
||||
ProcessCurrentHandlerRefreshTask task = claimNextRunnableTaskUnderDispatchLock();
|
||||
if (task == null) {
|
||||
return;
|
||||
}
|
||||
String workerId = IdUtil.fastSimpleUUID();
|
||||
refreshWorkerLease(workerId);
|
||||
log.info("当前处理人刷新任务派工成功，workerId：{}，activeWorkers：{}，maxWorkers：{}，流程实例id：{}，{}",
|
||||
workerId, activeWorkerCount, refreshProperties.getMaxWorkers(), task.getProcessInstanceId(), formatTaskLog(task));
|
||||
// worker租约到期前再触发一次派工，用于兜底恢复异常中断任务
|
||||
scheduleDispatch(workerLeaseMillis());
|
||||
asyncService.execute(() -> workerLoop(workerId, task));
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("当前处理人刷新派工被中断", e);
|
||||
} catch (Exception e) {
|
||||
log.error("当前处理人刷新派工异常", e);
|
||||
} finally {
|
||||
unlock(dispatchLock);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* worker循环拉取任务，尽量复用已经占用的worker槽位
|
||||
*
|
||||
* @param workerId worker id
|
||||
* @param firstTask 第一条任务
|
||||
*/
|
||||
private void workerLoop(String workerId, ProcessCurrentHandlerRefreshTask firstTask) {
|
||||
try {
|
||||
ProcessCurrentHandlerRefreshTask task = firstTask;
|
||||
while (task != null) {
|
||||
log.info("当前处理人刷新worker开始执行任务，workerId：{}，流程实例id：{}，{}", workerId, task.getProcessInstanceId(), formatTaskLog(task));
|
||||
processTask(workerId, task);
|
||||
refreshWorkerLease(workerId);
|
||||
task = claimNextRunnableTask();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("当前处理人刷新worker执行异常，workerId：{}", workerId, e);
|
||||
} finally {
|
||||
removeWorkerLease(workerId);
|
||||
log.info("当前处理人刷新worker结束，workerId：{}", workerId);
|
||||
tryDispatch();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行单条刷新任务
|
||||
*
|
||||
* @param workerId worker id
|
||||
* @param task 任务
|
||||
*/
|
||||
private void processTask(String workerId, ProcessCurrentHandlerRefreshTask task) {
|
||||
String processInstanceId = task.getProcessInstanceId();
|
||||
String runToken = task.getRunToken();
|
||||
if (!isTaskTokenMatched(processInstanceId, runToken)) {
|
||||
log.info("当前处理人刷新任务执行前token已失效，workerId：{}，流程实例id：{}，runToken：{}", workerId, processInstanceId, runToken);
|
||||
return;
|
||||
}
|
||||
updateHeartbeat(processInstanceId, runToken);
|
||||
long startAt = System.currentTimeMillis();
|
||||
log.info("当前处理人刷新任务开始查询，workerId：{}，流程实例id：{}，runToken：{}，attemptCount：{}，baselineSnapshot：{}",
|
||||
workerId, processInstanceId, runToken, task.getAttemptCount(), task.getBaselineSnapshot());
|
||||
FR<BusinessProcessVO> result = processClient.refreshBusinessProcessCurrentHandlers(buildUpdateParam(task.getContext()));
|
||||
updateHeartbeat(processInstanceId, runToken);
|
||||
if (!isTaskTokenMatched(processInstanceId, runToken)) {
|
||||
log.info("当前处理人刷新任务查询后token已失效，workerId：{}，流程实例id：{}，runToken：{}", workerId, processInstanceId, runToken);
|
||||
return;
|
||||
}
|
||||
if (result == null || FR.isNotSuccess(result)) {
|
||||
log.error("刷新当前处理人失败，流程实例id：{} result：{}", processInstanceId, JSON.toJSONString(result));
|
||||
requeueAfterMiss(task, false);
|
||||
return;
|
||||
}
|
||||
BusinessProcessVO businessProcessVO = result.getData();
|
||||
// 只有“提交”事件需要额外等待离开回调节点；
|
||||
// 审批通过/会签等场景允许节点不变但处理人变化，不能套用同一条规则，否则会误判为一直未流转
|
||||
if (shouldWaitForNextNode(task.getContext(), businessProcessVO)) {
|
||||
log.info("当前处理人刷新任务命中等待下一节点条件，workerId：{}，流程实例id：{}，耗时：{}ms，latestSnapshot：{}",
|
||||
workerId, processInstanceId, System.currentTimeMillis() - startAt, buildSnapshot(businessProcessVO));
|
||||
requeueAfterMiss(task, false);
|
||||
return;
|
||||
}
|
||||
String latestSnapshot = buildSnapshot(businessProcessVO);
|
||||
if (StringUtils.equals(latestSnapshot, task.getBaselineSnapshot())) {
|
||||
log.info("当前处理人刷新任务快照未变化，workerId：{}，流程实例id：{}，耗时：{}ms，baselineSnapshot：{}，latestSnapshot：{}",
|
||||
workerId, processInstanceId, System.currentTimeMillis() - startAt, task.getBaselineSnapshot(), latestSnapshot);
|
||||
requeueAfterMiss(task, false);
|
||||
return;
|
||||
}
|
||||
log.info("当前处理人刷新任务命中成功条件，workerId：{}，流程实例id：{}，耗时：{}ms，baselineSnapshot：{}，latestSnapshot：{}",
|
||||
workerId, processInstanceId, System.currentTimeMillis() - startAt, task.getBaselineSnapshot(), latestSnapshot);
|
||||
handleRefreshSuccess(task, businessProcessVO, latestSnapshot);
|
||||
refreshWorkerLease(workerId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理刷新成功
|
||||
*
|
||||
* @param task 任务
|
||||
* @param businessProcessVO 最新流程快照
|
||||
* @param latestSnapshot 最新快照
|
||||
*/
|
||||
private void handleRefreshSuccess(ProcessCurrentHandlerRefreshTask task, BusinessProcessVO businessProcessVO, String latestSnapshot) {
|
||||
String processInstanceId = task.getProcessInstanceId();
|
||||
String runToken = task.getRunToken();
|
||||
AbstractProcessOperationHandler handler = getHandler(task.getProcessType());
|
||||
if (handler == null) {
|
||||
log.error("未找到当前处理人刷新处理器，流程类型：{} 流程实例id：{}", task.getProcessType(), processInstanceId);
|
||||
requeueAfterMiss(task, true);
|
||||
return;
|
||||
}
|
||||
ProcessOperationContext callbackParam = task.getContext();
|
||||
try {
|
||||
handler.handleCurrentHandlerRefresh(callbackParam, businessProcessVO);
|
||||
} catch (Exception e) {
|
||||
log.error("刷新当前处理人后执行业务回调异常，流程实例id：{}", processInstanceId, e);
|
||||
requeueAfterMiss(task, true);
|
||||
return;
|
||||
}
|
||||
|
||||
RLock lock = getProcessLock(processInstanceId);
|
||||
boolean locked = false;
|
||||
try {
|
||||
locked = lock.tryLock(refreshProperties.getLockWaitSeconds(), TimeUnit.SECONDS);
|
||||
if (!locked) {
|
||||
log.warn("刷新成功后回写任务失败，未获取到流程锁，流程实例id：{}", processInstanceId);
|
||||
scheduleDispatch(intervalMillis());
|
||||
return;
|
||||
}
|
||||
ProcessCurrentHandlerRefreshTask latestTask = getTask(processInstanceId);
|
||||
if (latestTask == null || !StringUtils.equals(runToken, latestTask.getRunToken())) {
|
||||
return;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
latestTask.setLatestSnapshot(latestSnapshot);
|
||||
latestTask.setLastSuccessAt(now);
|
||||
latestTask.setStartedAt(null);
|
||||
latestTask.setHeartbeatAt(null);
|
||||
latestTask.setRunToken(null);
|
||||
if (latestTask.getDesiredVersion() > task.getProcessingVersion()) {
|
||||
// 有新版本到来时，把最新快照提升为新基线，并重新排队到后面，避免一个流程长期占用worker
|
||||
latestTask.setBaselineSnapshot(latestSnapshot);
|
||||
latestTask.setAttemptCount(0);
|
||||
latestTask.setState(TaskState.STATE_WAITING);
|
||||
latestTask.setNextRetryAt(now + intervalMillis());
|
||||
saveTask(latestTask);
|
||||
putWaitingTask(processInstanceId, latestTask.getNextRetryAt());
|
||||
log.info("当前处理人刷新任务成功后发现新版本，重新排队，流程实例id：{}，{}", processInstanceId, formatTaskLog(latestTask));
|
||||
scheduleDispatch(intervalMillis());
|
||||
return;
|
||||
}
|
||||
latestTask.setState(TaskState.STATE_DONE);
|
||||
saveTask(latestTask, Duration.ofMinutes(refreshProperties.getDoneTtlMinutes()));
|
||||
removeWaitingTask(processInstanceId);
|
||||
log.info("当前处理人刷新任务执行完成，流程实例id：{}，{}", processInstanceId, formatTaskLog(latestTask));
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("刷新成功后回写任务被中断，流程实例id：{}", processInstanceId, e);
|
||||
} finally {
|
||||
unlock(lock);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 未命中最新快照时重新排队
|
||||
*
|
||||
* @param task 任务
|
||||
* @param resetAttempt 是否重置重试次数
|
||||
*/
|
||||
private void requeueAfterMiss(ProcessCurrentHandlerRefreshTask task, boolean resetAttempt) {
|
||||
String processInstanceId = task.getProcessInstanceId();
|
||||
String runToken = task.getRunToken();
|
||||
RLock lock = getProcessLock(processInstanceId);
|
||||
boolean locked = false;
|
||||
try {
|
||||
locked = lock.tryLock(refreshProperties.getLockWaitSeconds(), TimeUnit.SECONDS);
|
||||
if (!locked) {
|
||||
log.warn("刷新任务重新排队失败，未获取到流程锁，流程实例id：{}", processInstanceId);
|
||||
scheduleDispatch(intervalMillis());
|
||||
return;
|
||||
}
|
||||
ProcessCurrentHandlerRefreshTask latestTask = getTask(processInstanceId);
|
||||
if (latestTask == null || !StringUtils.equals(runToken, latestTask.getRunToken())) {
|
||||
return;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
boolean hasNewVersion = latestTask.getDesiredVersion() > task.getProcessingVersion();
|
||||
latestTask.setRunToken(null);
|
||||
latestTask.setStartedAt(null);
|
||||
latestTask.setHeartbeatAt(null);
|
||||
latestTask.setState(TaskState.STATE_WAITING);
|
||||
latestTask.setNextRetryAt(now + intervalMillis());
|
||||
if (resetAttempt || hasNewVersion) {
|
||||
latestTask.setAttemptCount(0);
|
||||
} else {
|
||||
latestTask.setAttemptCount(latestTask.getAttemptCount() + 1);
|
||||
}
|
||||
if (latestTask.getAttemptCount() >= refreshProperties.getMaxAttempts()) {
|
||||
latestTask.setState(TaskState.STATE_FAILED);
|
||||
saveTask(latestTask, Duration.ofMinutes(refreshProperties.getFailedTtlMinutes()));
|
||||
removeWaitingTask(processInstanceId);
|
||||
log.warn("当前处理人刷新任务达到最大重试次数，流程实例id：{}，{}", processInstanceId, formatTaskLog(latestTask));
|
||||
return;
|
||||
}
|
||||
saveTask(latestTask);
|
||||
putWaitingTask(processInstanceId, latestTask.getNextRetryAt());
|
||||
log.info("当前处理人刷新任务重新排队，流程实例id：{}，resetAttempt：{}，hasNewVersion：{}，{}",
|
||||
processInstanceId, resetAttempt, hasNewVersion, formatTaskLog(latestTask));
|
||||
scheduleDispatch(intervalMillis());
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("刷新任务重新排队被中断，流程实例id：{}", processInstanceId, e);
|
||||
} finally {
|
||||
unlock(lock);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* claim下一条可执行任务
|
||||
*
|
||||
* @return 任务，不存在时返回null
|
||||
*/
|
||||
private ProcessCurrentHandlerRefreshTask claimNextRunnableTask() {
|
||||
RLock dispatchLock = getDispatchLock();
|
||||
boolean locked = false;
|
||||
try {
|
||||
locked = dispatchLock.tryLock(refreshProperties.getLockWaitSeconds(), TimeUnit.SECONDS);
|
||||
if (!locked) {
|
||||
return null;
|
||||
}
|
||||
recoverTimeoutTasks();
|
||||
return claimNextRunnableTaskUnderDispatchLock();
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("claim当前处理人刷新任务被中断", e);
|
||||
return null;
|
||||
} finally {
|
||||
unlock(dispatchLock);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 在已持有派工锁的前提下claim下一条可执行任务
|
||||
*
|
||||
* @return 任务，不存在时返回null
|
||||
*/
|
||||
private ProcessCurrentHandlerRefreshTask claimNextRunnableTaskUnderDispatchLock() {
|
||||
Set<String> processIds = bladeRedis.getStringRedisTemplate().opsForZSet()
|
||||
.rangeByScore(ProcessLockKeyConstant.WAITING_KEY, 0, System.currentTimeMillis(), 0, 1);
|
||||
if (processIds == null || processIds.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
String processInstanceId = processIds.iterator().next();
|
||||
RLock processLock = getProcessLock(processInstanceId);
|
||||
boolean processLocked = false;
|
||||
try {
|
||||
processLocked = processLock.tryLock(refreshProperties.getLockWaitSeconds(), TimeUnit.SECONDS);
|
||||
if (!processLocked) {
|
||||
return null;
|
||||
}
|
||||
ProcessCurrentHandlerRefreshTask task = getTask(processInstanceId);
|
||||
if (task == null) {
|
||||
removeWaitingTask(processInstanceId);
|
||||
return null;
|
||||
}
|
||||
if (!TaskState.STATE_WAITING.equals(task.getState())) {
|
||||
removeWaitingTask(processInstanceId);
|
||||
return null;
|
||||
}
|
||||
if (task.getNextRetryAt() > System.currentTimeMillis()) {
|
||||
putWaitingTask(processInstanceId, task.getNextRetryAt());
|
||||
return null;
|
||||
}
|
||||
task.setState(TaskState.STATE_RUNNING);
|
||||
task.setProcessingVersion(task.getDesiredVersion());
|
||||
task.setRunToken(IdUtil.fastSimpleUUID());
|
||||
task.setStartedAt(System.currentTimeMillis());
|
||||
task.setHeartbeatAt(task.getStartedAt());
|
||||
saveTask(task);
|
||||
removeWaitingTask(processInstanceId);
|
||||
log.info("当前处理人刷新任务claim成功，流程实例id：{}，{}", processInstanceId, formatTaskLog(task));
|
||||
return task;
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("claim当前处理人刷新任务被中断，流程实例id：{}", processInstanceId, e);
|
||||
return null;
|
||||
} finally {
|
||||
if (processLocked) {
|
||||
unlock(processLock);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 恢复超时的运行中任务
|
||||
*/
|
||||
private void recoverTimeoutTasks() {
|
||||
Set<String> taskKeys = bladeRedis.getStringRedisTemplate().keys(ProcessLockKeyConstant.TASK_KEY_PREFIX + "*");
|
||||
if (taskKeys == null || taskKeys.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
for (String taskKey : taskKeys) {
|
||||
ProcessCurrentHandlerRefreshTask task = getTaskByKey(taskKey);
|
||||
if (task == null || !TaskState.STATE_RUNNING.equals(task.getState())) {
|
||||
continue;
|
||||
}
|
||||
Long heartbeatAt = task.getHeartbeatAt();
|
||||
if (heartbeatAt != null && now - heartbeatAt <= workerLeaseMillis()) {
|
||||
continue;
|
||||
}
|
||||
String processInstanceId = task.getProcessInstanceId();
|
||||
RLock lock = getProcessLock(processInstanceId);
|
||||
boolean locked = false;
|
||||
try {
|
||||
locked = lock.tryLock(refreshProperties.getLockWaitSeconds(), TimeUnit.SECONDS);
|
||||
if (!locked) {
|
||||
continue;
|
||||
}
|
||||
ProcessCurrentHandlerRefreshTask latestTask = getTask(processInstanceId);
|
||||
if (latestTask == null || !TaskState.STATE_RUNNING.equals(latestTask.getState())) {
|
||||
continue;
|
||||
}
|
||||
Long latestHeartbeatAt = latestTask.getHeartbeatAt();
|
||||
if (latestHeartbeatAt != null && now - latestHeartbeatAt <= workerLeaseMillis()) {
|
||||
continue;
|
||||
}
|
||||
latestTask.setState(TaskState.STATE_WAITING);
|
||||
latestTask.setRunToken(null);
|
||||
latestTask.setStartedAt(null);
|
||||
latestTask.setHeartbeatAt(null);
|
||||
latestTask.setNextRetryAt(now);
|
||||
saveTask(latestTask);
|
||||
putWaitingTask(processInstanceId, now);
|
||||
log.warn("恢复超时的当前处理人刷新任务，流程实例id：{}，{}", processInstanceId, formatTaskLog(latestTask));
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("恢复超时任务被中断，流程实例id：{}", processInstanceId, e);
|
||||
return;
|
||||
} finally {
|
||||
unlock(lock);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 查询业务流程当前快照
|
||||
*
|
||||
* @param processInstanceId 流程实例id
|
||||
* @return 快照字符串
|
||||
*/
|
||||
private String queryCurrentSnapshot(String processInstanceId) {
|
||||
FR<BusinessProcessVO> result = processClient.queryBusinessProcessSnapshot(processInstanceId);
|
||||
if (result == null || FR.isNotSuccess(result)) {
|
||||
log.warn("查询业务流程当前快照失败，流程实例id：{} result：{}", processInstanceId, JSON.toJSONString(result));
|
||||
return buildSnapshot(null);
|
||||
}
|
||||
return buildSnapshot(result.getData());
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造刷新请求参数
|
||||
*
|
||||
* @param callbackParam 回调参数
|
||||
* @return 刷新参数
|
||||
*/
|
||||
private BusinessProcessCurrentHandlerRefreshDTO buildUpdateParam(ProcessOperationContext callbackParam) {
|
||||
BusinessProcessCurrentHandlerRefreshDTO updateParam = new BusinessProcessCurrentHandlerRefreshDTO();
|
||||
updateParam.setProcessInstanceId(callbackParam.getProcessInstanceId());
|
||||
updateParam.setPromoterLoginName(callbackParam.getApplicantLoginName());
|
||||
updateParam.setComplete(callbackParam.isComplete());
|
||||
return updateParam;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造快照，统一用当前节点+当前处理人作为变更依据
|
||||
*
|
||||
* @param businessProcessVO 业务流程快照
|
||||
* @return 快照字符串
|
||||
*/
|
||||
private String buildSnapshot(BusinessProcessVO businessProcessVO) {
|
||||
if (businessProcessVO == null) {
|
||||
return "|";
|
||||
}
|
||||
return normalizeCsv(businessProcessVO.getCurrentNodeIds()) + "|" + normalizeCsv(businessProcessVO.getCurrentHandlers());
|
||||
}
|
||||
|
||||
/**
|
||||
* 提交回调时，如果刷新后仍停留在本次回调节点，说明流程尚未真正流转到下一激活节点，需要继续等待。
|
||||
* <p>
|
||||
* 这里只针对提交事件生效，不能推广到审批通过/会签等场景：
|
||||
* 会签节点在部分人审批完成后，当前节点可能仍然不变，但当前处理人已经发生变化，
|
||||
* 此时应当允许按“快照变化”判定成功，而不是继续等待节点变化。
|
||||
* </p>
|
||||
*
|
||||
* @param callbackParam 回调参数
|
||||
* @param businessProcessVO 最新流程快照
|
||||
* @return 是否继续等待下一节点
|
||||
*/
|
||||
private boolean shouldWaitForNextNode(ProcessOperationContext callbackParam, BusinessProcessVO businessProcessVO) {
|
||||
if (callbackParam == null || businessProcessVO == null) {
|
||||
return false;
|
||||
}
|
||||
if (ProcessCallbackType.SUBMIT != ProcessCallbackType.getCallbackType(callbackParam.getOperation())) {
|
||||
return false;
|
||||
}
|
||||
String callbackNodeId = callbackParam.getCurrentNodeId();
|
||||
if (StringUtils.isBlank(callbackNodeId)) {
|
||||
return false;
|
||||
}
|
||||
return containsCsvValue(businessProcessVO.getCurrentNodeIds(), callbackNodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 统一规范逗号拼接字段，避免比较时顺序影响结果
|
||||
*
|
||||
* @param value 原始值
|
||||
* @return 规范化后的字符串
|
||||
*/
|
||||
private String normalizeCsv(String value) {
|
||||
if (StringUtils.isBlank(value)) {
|
||||
return "";
|
||||
}
|
||||
return Stream.of(value.split(","))
|
||||
.map(String::trim)
|
||||
.filter(StringUtils::isNotBlank)
|
||||
.distinct()
|
||||
.sorted()
|
||||
.collect(Collectors.joining(","));
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断逗号分隔字段中是否包含指定值
|
||||
*
|
||||
* @param csv 逗号分隔字段
|
||||
* @param target 目标值
|
||||
* @return 是否包含
|
||||
*/
|
||||
private boolean containsCsvValue(String csv, String target) {
|
||||
if (StringUtils.isAnyBlank(csv, target)) {
|
||||
return false;
|
||||
}
|
||||
return Stream.of(csv.split(","))
|
||||
.map(String::trim)
|
||||
.anyMatch(target::equals);
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断任务token是否仍然有效
|
||||
*
|
||||
* @param processInstanceId 流程实例id
|
||||
* @param runToken 运行token
|
||||
* @return 是否匹配
|
||||
*/
|
||||
private boolean isTaskTokenMatched(String processInstanceId, String runToken) {
|
||||
ProcessCurrentHandlerRefreshTask latestTask = getTask(processInstanceId);
|
||||
return latestTask != null
|
||||
&& TaskState.STATE_RUNNING.equals(latestTask.getState())
|
||||
&& StringUtils.equals(runToken, latestTask.getRunToken());
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新任务心跳，表示当前worker仍然存活
|
||||
*
|
||||
* @param processInstanceId 流程实例id
|
||||
* @param runToken 运行token
|
||||
*/
|
||||
private void updateHeartbeat(String processInstanceId, String runToken) {
|
||||
RLock lock = getProcessLock(processInstanceId);
|
||||
boolean locked = false;
|
||||
try {
|
||||
locked = lock.tryLock(refreshProperties.getLockWaitSeconds(), TimeUnit.SECONDS);
|
||||
if (!locked) {
|
||||
return;
|
||||
}
|
||||
ProcessCurrentHandlerRefreshTask task = getTask(processInstanceId);
|
||||
if (task == null || !StringUtils.equals(runToken, task.getRunToken())) {
|
||||
return;
|
||||
}
|
||||
task.setHeartbeatAt(System.currentTimeMillis());
|
||||
saveTask(task);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("更新刷新任务心跳被中断，流程实例id：{}", processInstanceId, e);
|
||||
} finally {
|
||||
unlock(lock);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 按默认方式保存任务
|
||||
*
|
||||
* @param task 任务
|
||||
*/
|
||||
private void saveTask(ProcessCurrentHandlerRefreshTask task) {
|
||||
bladeRedis.getStringRedisTemplate().opsForValue().set(getTaskKey(task.getProcessInstanceId()), JSON.toJSONString(task));
|
||||
}
|
||||
|
||||
/**
|
||||
* 按TTL保存任务
|
||||
*
|
||||
* @param task 任务
|
||||
* @param ttl TTL
|
||||
*/
|
||||
private void saveTask(ProcessCurrentHandlerRefreshTask task, Duration ttl) {
|
||||
bladeRedis.getStringRedisTemplate().opsForValue()
|
||||
.set(getTaskKey(task.getProcessInstanceId()), JSON.toJSONString(task), ttl);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取任务
|
||||
*
|
||||
* @param processInstanceId 流程实例id
|
||||
* @return 任务
|
||||
*/
|
||||
private ProcessCurrentHandlerRefreshTask getTask(String processInstanceId) {
|
||||
return getTaskByKey(getTaskKey(processInstanceId));
|
||||
}
|
||||
|
||||
/**
|
||||
* 通过key读取任务
|
||||
*
|
||||
* @param taskKey 任务key
|
||||
* @return 任务
|
||||
*/
|
||||
private ProcessCurrentHandlerRefreshTask getTaskByKey(String taskKey) {
|
||||
String content = bladeRedis.getStringRedisTemplate().opsForValue().get(taskKey);
|
||||
if (StringUtils.isBlank(content)) {
|
||||
return null;
|
||||
}
|
||||
return JSON.parseObject(content, ProcessCurrentHandlerRefreshTask.class);
|
||||
}
|
||||
|
||||
/**
|
||||
* 放入等待队列
|
||||
*
|
||||
* @param processInstanceId 流程实例id
|
||||
* @param nextRetryAt 下次执行时间
|
||||
*/
|
||||
private void putWaitingTask(String processInstanceId, long nextRetryAt) {
|
||||
bladeRedis.getStringRedisTemplate().opsForZSet().add(ProcessLockKeyConstant.WAITING_KEY, processInstanceId, nextRetryAt);
|
||||
}
|
||||
|
||||
/**
|
||||
* 移除等待队列中的任务
|
||||
*
|
||||
* @param processInstanceId 流程实例id
|
||||
*/
|
||||
private void removeWaitingTask(String processInstanceId) {
|
||||
bladeRedis.getStringRedisTemplate().opsForZSet().remove(ProcessLockKeyConstant.WAITING_KEY, processInstanceId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 是否存在等待任务
|
||||
*
|
||||
* @return 是否存在
|
||||
*/
|
||||
private boolean hasWaitingTask() {
|
||||
Long size = bladeRedis.getStringRedisTemplate().opsForZSet().zCard(ProcessLockKeyConstant.WAITING_KEY);
|
||||
return size != null && size > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取worker租约map
|
||||
*
|
||||
* @return worker租约map
|
||||
*/
|
||||
private RMapCache<String, String> getWorkerLeaseMap() {
|
||||
return redisLockClient.getRedissonClient().getMapCache(ProcessLockKeyConstant.WORKER_LEASE_KEY);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取当前活跃worker数量
|
||||
*
|
||||
* @return 活跃worker数量
|
||||
*/
|
||||
private int getActiveWorkerCount() {
|
||||
return getWorkerLeaseMap().size();
|
||||
}
|
||||
|
||||
/**
|
||||
* 刷新worker租约
|
||||
*
|
||||
* @param workerId worker id
|
||||
*/
|
||||
private void refreshWorkerLease(String workerId) {
|
||||
getWorkerLeaseMap().put(workerId, workerId, refreshProperties.getWorkerLeaseSeconds(), TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除worker租约
|
||||
*
|
||||
* @param workerId worker id
|
||||
*/
|
||||
private void removeWorkerLease(String workerId) {
|
||||
getWorkerLeaseMap().remove(workerId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 安排稍后再次派工
|
||||
*
|
||||
* @param delayMillis 延迟毫秒数
|
||||
*/
|
||||
private void scheduleDispatch(long delayMillis) {
|
||||
asyncService.delayExecute(delayMillis, this::tryDispatch);
|
||||
}
|
||||
|
||||
private RLock getProcessLock(String processInstanceId) {
|
||||
return redisLockClient.getRedissonClient().getLock(ProcessLockKeyConstant.PROCESS_LOCK_KEY_PREFIX + processInstanceId);
|
||||
}
|
||||
|
||||
private RLock getDispatchLock() {
|
||||
return redisLockClient.getRedissonClient().getLock(ProcessLockKeyConstant.DISPATCH_LOCK_KEY);
|
||||
}
|
||||
|
||||
/**
|
||||
* 释放锁
|
||||
*
|
||||
* @param lock
|
||||
*/
|
||||
private void unlock(RLock lock) {
|
||||
if (lock.isLocked() && lock.isHeldByCurrentThread()) {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private String getTaskKey(String processInstanceId) {
|
||||
return ProcessLockKeyConstant.TASK_KEY_PREFIX + processInstanceId;
|
||||
}
|
||||
|
||||
private long initialDelayMillis() {
|
||||
return refreshProperties.getInitialDelaySeconds() * 1000L;
|
||||
}
|
||||
|
||||
private long intervalMillis() {
|
||||
return refreshProperties.getIntervalSeconds() * 1000L;
|
||||
}
|
||||
|
||||
private long workerLeaseMillis() {
|
||||
return refreshProperties.getWorkerLeaseSeconds() * 1000L;
|
||||
}
|
||||
|
||||
private String formatTaskLog(ProcessCurrentHandlerRefreshTask task) {
|
||||
if (task == null) {
|
||||
return "task=null";
|
||||
}
|
||||
return "state=" + task.getState()
|
||||
+ ", desiredVersion=" + task.getDesiredVersion()
|
||||
+ ", processingVersion=" + task.getProcessingVersion()
|
||||
+ ", attemptCount=" + task.getAttemptCount()
|
||||
+ ", nextRetryAt=" + task.getNextRetryAt()
|
||||
+ ", startedAt=" + task.getStartedAt()
|
||||
+ ", heartbeatAt=" + task.getHeartbeatAt()
|
||||
+ ", lastCallbackAt=" + task.getLastCallbackAt()
|
||||
+ ", lastSuccessAt=" + task.getLastSuccessAt();
|
||||
}
|
||||
}
|
||||
+80
@@ -0,0 +1,80 @@
|
||||
package org.springblade.openapi.mk.support.handler;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springblade.openapi.mk.support.base.ProcessOperationContext;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* 当前处理人刷新任务
|
||||
*
|
||||
* @author bfhuange
|
||||
* @date 2026/4/9
|
||||
*/
|
||||
@Data
|
||||
public class ProcessCurrentHandlerRefreshTask implements Serializable {
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
/**
|
||||
* 流程实例id
|
||||
*/
|
||||
private String processInstanceId;
|
||||
/**
|
||||
* 流程类型
|
||||
*/
|
||||
private String processType;
|
||||
/**
|
||||
* 任务状态
|
||||
*/
|
||||
private String state;
|
||||
/**
|
||||
* 期望处理版本
|
||||
*/
|
||||
private long desiredVersion;
|
||||
/**
|
||||
* 当前执行版本
|
||||
*/
|
||||
private long processingVersion;
|
||||
/**
|
||||
* 当前执行令牌
|
||||
*/
|
||||
private String runToken;
|
||||
/**
|
||||
* 当前基线快照
|
||||
*/
|
||||
private String baselineSnapshot;
|
||||
/**
|
||||
* 最近一次成功快照
|
||||
*/
|
||||
private String latestSnapshot;
|
||||
/**
|
||||
* 重试次数
|
||||
*/
|
||||
private int attemptCount;
|
||||
/**
|
||||
* 下次重试时间
|
||||
*/
|
||||
private long nextRetryAt;
|
||||
/**
|
||||
* 开始执行时间
|
||||
*/
|
||||
private Long startedAt;
|
||||
/**
|
||||
* 最近心跳时间
|
||||
*/
|
||||
private Long heartbeatAt;
|
||||
/**
|
||||
* 最近成功时间
|
||||
*/
|
||||
private Long lastSuccessAt;
|
||||
/**
|
||||
* 最近回调时间
|
||||
*/
|
||||
private Long lastCallbackAt;
|
||||
/**
|
||||
* 最近一次回调参数
|
||||
*/
|
||||
private ProcessOperationContext context;
|
||||
}
|
||||
+25
@@ -0,0 +1,25 @@
|
||||
package org.springblade.openapi.mk.support.handler;
|
||||
|
||||
/**
|
||||
* 当前处理人刷新任务状态常量类
|
||||
* @author bfhuange
|
||||
* @since 2026/4/9
|
||||
*/
|
||||
public class TaskState {
|
||||
/**
|
||||
* 等待执行
|
||||
*/
|
||||
public static final String STATE_WAITING = "WAITING";
|
||||
/**
|
||||
* 执行中
|
||||
*/
|
||||
public static final String STATE_RUNNING = "RUNNING";
|
||||
/**
|
||||
* 执行完成
|
||||
*/
|
||||
public static final String STATE_DONE = "DONE";
|
||||
/**
|
||||
* 执行失败
|
||||
*/
|
||||
public static final String STATE_FAILED = "FAILED";
|
||||
}
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
package org.springblade.openapi.mk.util;
|
||||
|
||||
import org.springblade.core.tool.utils.StringUtil;
|
||||
|
||||
/**
|
||||
* @author bfhuange
|
||||
* @since 2026/7/12
|
||||
*/
|
||||
public class ProcessTypeUtils {
|
||||
|
||||
/**
|
||||
* 获取流程类型
|
||||
* @param templateCode
|
||||
* @param templateCodePrefix
|
||||
* @return
|
||||
*/
|
||||
public static String getProcessType(String templateCode, String templateCodePrefix) {
|
||||
if (StringUtil.isBlank(templateCode)) {
|
||||
return templateCode;
|
||||
}
|
||||
return templateCode.replace(templateCodePrefix, "");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
server:
|
||||
port: 8108
|
||||
|
||||
spring:
|
||||
application:
|
||||
name: blade-openapi
|
||||
config:
|
||||
import:
|
||||
- nacos:blade.yaml?group=DEFAULT_GROUP&refreshEnabled=true
|
||||
- nacos:blade-${spring.profiles.active}.yaml?group=DEFAULT_GROUP&refreshEnabled=true
|
||||
- nacos:third-party-api.yaml?group=DEFAULT_GROUP&refreshEnabled=true
|
||||
- nacos:blade-openapi-dynamictp.yaml?group=DEFAULT_GROUP&refreshEnabled=true
|
||||
- optional:nacos:${spring.application.name}-${spring.profiles.active}.yaml?group=DEFAULT_GROUP&refreshEnabled=true
|
||||
cloud:
|
||||
nacos:
|
||||
username: ${NACOS_USERNAME:nacos}
|
||||
password: ${NACOS_PASSWORD:nacos}
|
||||
server-addr: ${NACOS_HOST:127.0.0.1:8848}
|
||||
discovery:
|
||||
namespace: ${NACOS_NAMESPACE:${spring.profiles.active}}
|
||||
config:
|
||||
file-extension: yaml
|
||||
namespace: ${NACOS_NAMESPACE:${spring.profiles.active}}
|
||||
datasource:
|
||||
url: ${blade.datasource.${spring.profiles.active}.url}
|
||||
username: ${blade.datasource.${spring.profiles.active}.username}
|
||||
password: ${blade.datasource.${spring.profiles.active}.password}
|
||||
@@ -1,6 +0,0 @@
|
||||
#spring:
|
||||
# cloud:
|
||||
# nacos:
|
||||
# username: nacos
|
||||
# password: ${NACOS_PASSWORD:gr30wIs5%Hi7keQj}
|
||||
# server-addr: ${NACOS_ADDR:10.38.16.127:8848}
|
||||
@@ -1,6 +0,0 @@
|
||||
#spring:
|
||||
# cloud:
|
||||
# nacos:
|
||||
# username: nacos
|
||||
# password: rWrMrVTWyf%ekjuw
|
||||
# server-addr: ${NACOS_ADDR:192.168.0.242:8848}
|
||||
@@ -1,8 +0,0 @@
|
||||
#server:
|
||||
# port: 38108
|
||||
#spring:
|
||||
# cloud:
|
||||
# nacos:
|
||||
# username: nacos
|
||||
# password: gr30wIs5%Hi7keQj
|
||||
# server-addr: ${NACOS_ADDR:10.38.16.127:8848}
|
||||
@@ -1,34 +0,0 @@
|
||||
server:
|
||||
port: 8108
|
||||
|
||||
#数据源配置
|
||||
spring:
|
||||
datasource:
|
||||
url: ${blade.datasource.dev.url}
|
||||
username: ${blade.datasource.dev.username}
|
||||
password: ${blade.datasource.dev.password}
|
||||
|
||||
#spring:
|
||||
# application:
|
||||
# name: blade-openapi
|
||||
# cloud:
|
||||
# nacos:
|
||||
# discovery:
|
||||
# namespace: ${spring.profiles.active}
|
||||
# # 不注册到nacos
|
||||
# #registerEnabled: false
|
||||
# config:
|
||||
# # 文件后缀名
|
||||
# file-extension: yaml
|
||||
# namespace: ${spring.profiles.active}
|
||||
# shared-configs:
|
||||
# - data-id: blade.yaml
|
||||
# refresh: true
|
||||
# - data-id: blade-${spring.profiles.active}.yaml
|
||||
# refresh: true
|
||||
# - data-id: third-party-api.yaml
|
||||
# refresh: true
|
||||
# extension-configs:
|
||||
# - data-id: ${spring.application.name}-dynamictp.yaml
|
||||
# group: DEFAULT_GROUP
|
||||
# refresh: true # 必须配置，负责自动刷新不生效
|
||||
+1
-1
@@ -40,7 +40,7 @@ public class BusinessProcessClient implements IBusinessProcessClient {
|
||||
}
|
||||
|
||||
@Override
|
||||
public FR<Boolean> updateBusinessProcessStatus(@Validated @RequestBody BusinessProcessUpdateDTO param) {
|
||||
public FR<String> updateBusinessProcessStatus(@Validated @RequestBody BusinessProcessUpdateDTO param) {
|
||||
return FR.data(businessProcessService.updateBusinessProcessStatus(param));
|
||||
}
|
||||
|
||||
|
||||
+1
-1
@@ -30,7 +30,7 @@ public interface IBusinessProcessService extends IService<BusinessProcess> {
|
||||
* @param param
|
||||
* @return
|
||||
*/
|
||||
boolean updateBusinessProcessStatus(BusinessProcessUpdateDTO param);
|
||||
String updateBusinessProcessStatus(BusinessProcessUpdateDTO param);
|
||||
|
||||
/**
|
||||
* 修改业务流程审批人
|
||||
|
||||
+5
-4
@@ -116,7 +116,7 @@ public class BusinessProcessServiceImpl extends ServiceImpl<BusinessProcessMappe
|
||||
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
@Override
|
||||
public boolean updateBusinessProcessStatus(BusinessProcessUpdateDTO param) {
|
||||
public String updateBusinessProcessStatus(BusinessProcessUpdateDTO param) {
|
||||
AssertUtils.notNull(param, "参数不能为空");
|
||||
BusinessProcess businessProcess = this.getOne(Wrappers.<BusinessProcess>lambdaQuery()
|
||||
.eq(BusinessProcess::getProcessInstanceId, param.getProcessInstanceId())
|
||||
@@ -128,9 +128,10 @@ public class BusinessProcessServiceImpl extends ServiceImpl<BusinessProcessMappe
|
||||
BusinessProcess updateParam = new BusinessProcess();
|
||||
updateParam.setId(businessProcess.getId());
|
||||
updateParam.setApproveStatus(param.getApproveStatus());
|
||||
return this.updateById(updateParam);
|
||||
}
|
||||
return true;
|
||||
boolean update = this.updateById(updateParam);
|
||||
return update ? param.getApproveStatus() : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
-- 业务流程表
|
||||
CREATE TABLE `blade_business_process`
|
||||
(
|
||||
`id` bigint NOT NULL COMMENT '主键',
|
||||
`biz_id` bigint NOT NULL COMMENT '业务id',
|
||||
`process_instance_id` varchar(40) NOT NULL COMMENT '流程实例id',
|
||||
`process_type` varchar(40) NOT NULL COMMENT '流程类型',
|
||||
`doc_code` varchar(50) DEFAULT NULL COMMENT '文档编号',
|
||||
`subject` varchar(500) DEFAULT NULL COMMENT '标题',
|
||||
`promoter_id` bigint NOT NULL COMMENT '发起人id',
|
||||
`promoter_name` varchar(40) NOT NULL COMMENT '发起人名称',
|
||||
`promoter_login_name` varchar(40) DEFAULT NULL COMMENT '发起人登录名',
|
||||
`submit_time` datetime NOT NULL COMMENT '提交时间',
|
||||
`complete_time` datetime DEFAULT NULL COMMENT '完成时间',
|
||||
`current_node_ids` varchar(40) DEFAULT NULL COMMENT '当前节点id，多个用逗号拼接',
|
||||
`current_node_names` varchar(40) DEFAULT NULL COMMENT '当前节点名称，多个用逗号拼接',
|
||||
`current_handlers` varchar(255) DEFAULT NULL COMMENT '当前处理人，多个用逗号拼接',
|
||||
`receive_time` datetime DEFAULT NULL COMMENT '接收时间',
|
||||
`is_completed` tinyint(1) DEFAULT '0' COMMENT '是否已完成',
|
||||
`approve_status` varchar(20) DEFAULT NULL COMMENT '审批状态',
|
||||
`tenant_id` varchar(12) DEFAULT '000000' COMMENT '租户ID',
|
||||
`create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
|
||||
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最后更新时间',
|
||||
PRIMARY KEY (`id`),
|
||||
KEY `idx_blade_business_process_process_instance_id` (`process_instance_id`) USING BTREE,
|
||||
KEY `idx_blade_business_process_promoter_login_name` (`promoter_login_name`) USING BTREE
|
||||
) ENGINE=InnoDB COMMENT='业务流程关联表';
|
||||
|
||||
-- 新增mk client配置
|
||||
INSERT INTO blade_client (id,client_id,client_secret,resource_ids,`scope`,authorized_grant_types,web_server_redirect_uri,authorities,access_token_validity,refresh_token_validity,additional_information,autoapprove,create_user,create_dept,create_time,update_user,update_time,status,is_deleted) VALUES
|
||||
(1834798269409857538,'mk-oauth','mk_oauth_secret','','all','refresh_token,password,authorization_code,captcha,social,sms_code,register','http://localhost:2888/login','',604800,604800,NULL,'true',1123598821738675201,1828387593663762436,'2024-09-14 11:36:11',1123598821738675201,'2024-09-14 11:36:11',1,0);
|
||||
Reference in New Issue
Block a user