From 0b8aecc49b048566b0f5dbfbad9a72679591b3c4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E6=96=8C=E5=B3=B0?= Date: Sun, 9 Aug 2026 23:31:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(openapi):=20=E6=B7=BB=E5=8A=A0mk=E5=9B=9E?= =?UTF-8?q?=E8=B0=83=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../springblade/openapi/mk/api/IApi4MK.java | 27 - .../mk/constant/ProcessLockKeyConstant.java | 30 + .../mk/pojo/dto/Api4MKProcessApprovalDTO.java | 9 - .../mk/pojo/dto/ApiMKProcessFinishDTO.java | 34 - .../mk/pojo/dto/ProcessApprovalDTO.java | 40 - .../process/feign/IBusinessProcessClient.java | 4 +- .../process/pojo/dto/ProcessExecuteDTO.java | 1 - blade-service/blade-openapi/pom.xml | 4 + .../openapi/OpenApiApplication.java | 1 + .../org/springblade/openapi/mk/Api4MK.java | 93 ++ .../mk/config/AsyncExecutorProperties.java | 27 + .../CurrentHandlerRefreshProperties.java | 52 + .../base/AbstractProcessOperationHandler.java | 301 ++++++ .../mk/support/base/ProcessHandler.java | 27 + .../support/base/ProcessOperationContext.java | 139 +++ .../support/base/ProcessOperationHandler.java | 45 + .../mk/support/handler/AsyncService.java | 113 +++ .../ProcessCurrentHandlerRefreshService.java | 903 ++++++++++++++++++ .../ProcessCurrentHandlerRefreshTask.java | 80 ++ .../openapi/mk/support/handler/TaskState.java | 25 + .../openapi/mk/util/ProcessTypeUtils.java | 23 + .../src/main/resources/application.yml | 27 + .../src/main/resources/bootstrap-dev.yml | 6 - .../src/main/resources/bootstrap-prod.yml | 6 - .../src/main/resources/bootstrap-test.yml | 8 - .../src/main/resources/bootstrap.yml | 34 - .../process/feign/BusinessProcessClient.java | 2 +- .../service/IBusinessProcessService.java | 2 +- .../impl/BusinessProcessServiceImpl.java | 9 +- doc/sql/changelog/process-202608071800.sql | 31 + 30 files changed, 1930 insertions(+), 173 deletions(-) create mode 100644 blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/constant/ProcessLockKeyConstant.java delete mode 100644 blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/ApiMKProcessFinishDTO.java delete mode 100644 blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/ProcessApprovalDTO.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/Api4MK.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/config/AsyncExecutorProperties.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/config/CurrentHandlerRefreshProperties.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/AbstractProcessOperationHandler.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessHandler.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessOperationContext.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessOperationHandler.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/AsyncService.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/ProcessCurrentHandlerRefreshService.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/ProcessCurrentHandlerRefreshTask.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/TaskState.java create mode 100644 blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/util/ProcessTypeUtils.java create mode 100644 blade-service/blade-openapi/src/main/resources/application.yml delete mode 100644 blade-service/blade-openapi/src/main/resources/bootstrap-dev.yml delete mode 100644 blade-service/blade-openapi/src/main/resources/bootstrap-prod.yml delete mode 100644 blade-service/blade-openapi/src/main/resources/bootstrap-test.yml delete mode 100644 blade-service/blade-openapi/src/main/resources/bootstrap.yml create mode 100644 doc/sql/changelog/process-202608071800.sql diff --git a/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/api/IApi4MK.java b/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/api/IApi4MK.java index 6c8d09d..701128a 100644 --- a/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/api/IApi4MK.java +++ b/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/api/IApi4MK.java @@ -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 processFinishCallback(@RequestBody Api4MKProcessApprovalDTO param); - - /** - * 流程审批同意回调接口 - * @param param - * @return - */ - @PostMapping(PROCESS_APPROVAL_CALLBACK) - FR processApprovalCallback(@RequestBody Api4MKProcessApprovalDTO param); - - /** - * 流程审批拒绝回调接口 - * @param param - * @return - */ - @PostMapping(PROCESS_REJECT_CALLBACK) - FR processRejectCallback(@RequestBody Api4MKProcessApprovalDTO param); - - /** - * 流程撤销回调接口 - * @param param - * @return - */ - @PostMapping(PROCESS_REVOKE_CALLBACK) - FR processRevokeCallback(@RequestBody Api4MKProcessApprovalDTO param); } diff --git a/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/constant/ProcessLockKeyConstant.java b/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/constant/ProcessLockKeyConstant.java new file mode 100644 index 0000000..4fa3c13 --- /dev/null +++ b/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/constant/ProcessLockKeyConstant.java @@ -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"; +} diff --git a/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/Api4MKProcessApprovalDTO.java b/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/Api4MKProcessApprovalDTO.java index b71f71e..0941684 100644 --- a/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/Api4MKProcessApprovalDTO.java +++ b/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/Api4MKProcessApprovalDTO.java @@ -69,13 +69,4 @@ public class Api4MKProcessApprovalDTO implements Serializable { */ private String operatorLoginName; - //====================非mk回调参数，回调接口设置参数=================== - /** - * 是否流程已完成，非mk回调参数，回调接口设置参数 - */ - private boolean complete; - /** - * 审批状态，非mk回调参数，回调接口设置参数 - */ - private String approveStatus; } diff --git a/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/ApiMKProcessFinishDTO.java b/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/ApiMKProcessFinishDTO.java deleted file mode 100644 index 2e2b57f..0000000 --- a/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/ApiMKProcessFinishDTO.java +++ /dev/null @@ -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; -} diff --git a/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/ProcessApprovalDTO.java b/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/ProcessApprovalDTO.java deleted file mode 100644 index 2bed1af..0000000 --- a/blade-service-api/blade-open-api/src/main/java/org/springblade/openapi/mk/pojo/dto/ProcessApprovalDTO.java +++ /dev/null @@ -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; - - -} diff --git a/blade-service-api/blade-process-api/src/main/java/org/springblade/process/feign/IBusinessProcessClient.java b/blade-service-api/blade-process-api/src/main/java/org/springblade/process/feign/IBusinessProcessClient.java index 9218fe8..16cc18e 100644 --- a/blade-service-api/blade-process-api/src/main/java/org/springblade/process/feign/IBusinessProcessClient.java +++ b/blade-service-api/blade-process-api/src/main/java/org/springblade/process/feign/IBusinessProcessClient.java @@ -49,10 +49,10 @@ public interface IBusinessProcessClient { /** * 修改业务流程状态 * @param param - * @return + * @return 审批状态 */ @PostMapping(UPDATE_BUSINESS_PROCESS_STATUS) - FR updateBusinessProcessStatus(@Validated @RequestBody BusinessProcessUpdateDTO param); + FR updateBusinessProcessStatus(@Validated @RequestBody BusinessProcessUpdateDTO param); /** * 修改业务流程审批人 diff --git a/blade-service-api/blade-process-api/src/main/java/org/springblade/process/pojo/dto/ProcessExecuteDTO.java b/blade-service-api/blade-process-api/src/main/java/org/springblade/process/pojo/dto/ProcessExecuteDTO.java index 981167c..d3a3a8d 100644 --- a/blade-service-api/blade-process-api/src/main/java/org/springblade/process/pojo/dto/ProcessExecuteDTO.java +++ b/blade-service-api/blade-process-api/src/main/java/org/springblade/process/pojo/dto/ProcessExecuteDTO.java @@ -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; diff --git a/blade-service/blade-openapi/pom.xml b/blade-service/blade-openapi/pom.xml index b25a56b..e036291 100644 --- a/blade-service/blade-openapi/pom.xml +++ b/blade-service/blade-openapi/pom.xml @@ -42,6 +42,10 @@ org.springblade blade-mk-api + + org.springblade + blade-process-api + org.mapstruct diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/OpenApiApplication.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/OpenApiApplication.java index 9815c2e..e76743f 100644 --- a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/OpenApiApplication.java +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/OpenApiApplication.java @@ -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); } diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/Api4MK.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/Api4MK.java new file mode 100644 index 0000000..0f43c46 --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/Api4MK.java @@ -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 handlerMap; + + public Api4MK(MKProperties mkProperties, ObjectProvider> 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 processCommonCallback(Api4MKProcessApprovalDTO param) { + log.info("mk流程通用回调 操作名称：{} 参数：{}", ProcessOperationType.getOperationName(param.getOperation()), JSON.toJSONString(param)); + callback(param, ProcessHandler::approve); + return FR.status(true); + } + + @Override + public FR 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 consumer) { + String processType = ProcessTypeUtils.getProcessType(param.getTemplateCode(), mkProperties.getTemplateCodePrefix()); + ProcessHandler handler = getHandler(processType); + if (handler != null) { + consumer.accept(handler, param); + return; + } + log.warn("未配置流程类型对应的处理器 流程类型：{}", processType); + } +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/config/AsyncExecutorProperties.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/config/AsyncExecutorProperties.java new file mode 100644 index 0000000..e85fe3f --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/config/AsyncExecutorProperties.java @@ -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"; +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/config/CurrentHandlerRefreshProperties.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/config/CurrentHandlerRefreshProperties.java new file mode 100644 index 0000000..741f8f7 --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/config/CurrentHandlerRefreshProperties.java @@ -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; +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/AbstractProcessOperationHandler.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/AbstractProcessOperationHandler.java new file mode 100644 index 0000000..f4f6668 --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/AbstractProcessOperationHandler.java @@ -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 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 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 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) {}; +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessHandler.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessHandler.java new file mode 100644 index 0000000..f0b06d3 --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessHandler.java @@ -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 getProcessTypes(); + + /** + * 通用审批 + * @param param + */ + void approve(Api4MKProcessApprovalDTO param); + +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessOperationContext.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessOperationContext.java new file mode 100644 index 0000000..5bc9690 --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessOperationContext.java @@ -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; + +/** + * 流程回调内部处理上下文。 + *

+ * callbackParam 仅保留 MK 原始回调参数， + * 其余字段为 openapi 在处理过程中补充的上下文参数。 + *

+ *

+ * 设计目的： + * 1. 避免把内部推导字段继续堆到 MK 原始回调 DTO 上； + * 2. 对外保留原始回调对象，便于排查问题、记录日志和后续扩展； + * 3. 通过代理 getter 尽量兼容原来直接读取 DTO 字段的使用习惯，降低老流程和后续分支合并成本。 + *

+ * + * @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(); + } +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessOperationHandler.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessOperationHandler.java new file mode 100644 index 0000000..c914acd --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/base/ProcessOperationHandler.java @@ -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); +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/AsyncService.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/AsyncService.java new file mode 100644 index 0000000..87250c5 --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/AsyncService.java @@ -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); + } + + /** + * 延迟执行指定毫秒数。 + *

+ * 这里改为使用 ScheduledDtpExecutor 做真正的定时调度， + * 避免再通过线程池线程 sleep 的方式占用工作线程，导致真正的业务任务迟迟无法启动。 + *

+ * + * @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 手动包装任务。 + *

+ * 当前使用的 dynamic-tp 版本下，ScheduledDtpExecutor 对 taskWrapper 的透传存在缺口， + * 这里直接读取线程池上已生效的 wrappers，按框架默认增强链顺序主动包装一次， + * 这样既能复用现有配置，又避免手写 mdc 透传逻辑与框架实现产生偏差。 + *

+ */ + 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; + } +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/ProcessCurrentHandlerRefreshService.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/ProcessCurrentHandlerRefreshService.java new file mode 100644 index 0000000..057f541 --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/ProcessCurrentHandlerRefreshService.java @@ -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; + +/** + * 当前处理人刷新调度服务。 + *

+ * 背景： + * 流程引擎回调业务系统时，流程往往还没有真正流转到下一个激活节点， + * 此时立即查询当前节点/当前处理人，拿到的仍可能是上一节点的旧结果。 + * 因此这里不再依赖一次性的固定延迟，而是改成“按流程实例维度入队 + 固定间隔轮询刷新”的调度模型。 + *

+ *

+ * 整体流程： + * 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. 成功完成的任务进入完成态并短期保留，随后自动过期。 + *

+ *

+ * 集群与并发约束： + * 1. 流程实例级别使用分布式锁，保证同一流程实例的任务状态变更串行化； + * 2. 派工使用全局分布式锁，保证多个实例不会同时超发 worker； + * 3. 活跃 worker 数通过 Redis 租约控制，服务异常中断后，租约超时即可视为 worker 失活； + * 4. 运行中的任务会持续更新心跳，如果服务升级、中断或线程异常退出，超时恢复逻辑会把任务重新转回等待态； + * 5. 启动时不会全量恢复运行中任务，避免在集群环境中误伤其他实例上仍在执行的任务。 + *

+ *

+ * 成功判定规则： + * 不再区分终态/非终态，也不依赖回调里传入的 complete true/false 单独判定是否成功， + * 统一以“当前节点变化 + 当前处理人变化后的最新快照”是否相对基线发生变化作为刷新成功依据。 + *

+ * + * @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> handlersProvider; + private final Map handlerMap; + + public ProcessCurrentHandlerRefreshService(AsyncService asyncService, + BladeRedis bladeRedis, + RedisLockClient redisLockClient, + IBusinessProcessClient processClient, + CurrentHandlerRefreshProperties refreshProperties, + ObjectProvider> 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 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 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 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 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 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 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()); + } + + /** + * 提交回调时，如果刷新后仍停留在本次回调节点，说明流程尚未真正流转到下一激活节点，需要继续等待。 + *

+ * 这里只针对提交事件生效，不能推广到审批通过/会签等场景： + * 会签节点在部分人审批完成后，当前节点可能仍然不变，但当前处理人已经发生变化， + * 此时应当允许按“快照变化”判定成功，而不是继续等待节点变化。 + *

+ * + * @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 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(); + } +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/ProcessCurrentHandlerRefreshTask.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/ProcessCurrentHandlerRefreshTask.java new file mode 100644 index 0000000..90b08d2 --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/ProcessCurrentHandlerRefreshTask.java @@ -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; +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/TaskState.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/TaskState.java new file mode 100644 index 0000000..0b255df --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/support/handler/TaskState.java @@ -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"; +} diff --git a/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/util/ProcessTypeUtils.java b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/util/ProcessTypeUtils.java new file mode 100644 index 0000000..6eff965 --- /dev/null +++ b/blade-service/blade-openapi/src/main/java/org/springblade/openapi/mk/util/ProcessTypeUtils.java @@ -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, ""); + } +} diff --git a/blade-service/blade-openapi/src/main/resources/application.yml b/blade-service/blade-openapi/src/main/resources/application.yml new file mode 100644 index 0000000..1905b44 --- /dev/null +++ b/blade-service/blade-openapi/src/main/resources/application.yml @@ -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} diff --git a/blade-service/blade-openapi/src/main/resources/bootstrap-dev.yml b/blade-service/blade-openapi/src/main/resources/bootstrap-dev.yml deleted file mode 100644 index 303fc85..0000000 --- a/blade-service/blade-openapi/src/main/resources/bootstrap-dev.yml +++ /dev/null @@ -1,6 +0,0 @@ -#spring: -# cloud: -# nacos: -# username: nacos -# password: ${NACOS_PASSWORD:gr30wIs5%Hi7keQj} -# server-addr: ${NACOS_ADDR:10.38.16.127:8848} diff --git a/blade-service/blade-openapi/src/main/resources/bootstrap-prod.yml b/blade-service/blade-openapi/src/main/resources/bootstrap-prod.yml deleted file mode 100644 index 673f396..0000000 --- a/blade-service/blade-openapi/src/main/resources/bootstrap-prod.yml +++ /dev/null @@ -1,6 +0,0 @@ -#spring: -# cloud: -# nacos: -# username: nacos -# password: rWrMrVTWyf%ekjuw -# server-addr: ${NACOS_ADDR:192.168.0.242:8848} diff --git a/blade-service/blade-openapi/src/main/resources/bootstrap-test.yml b/blade-service/blade-openapi/src/main/resources/bootstrap-test.yml deleted file mode 100644 index d449d60..0000000 --- a/blade-service/blade-openapi/src/main/resources/bootstrap-test.yml +++ /dev/null @@ -1,8 +0,0 @@ -#server: -# port: 38108 -#spring: -# cloud: -# nacos: -# username: nacos -# password: gr30wIs5%Hi7keQj -# server-addr: ${NACOS_ADDR:10.38.16.127:8848} diff --git a/blade-service/blade-openapi/src/main/resources/bootstrap.yml b/blade-service/blade-openapi/src/main/resources/bootstrap.yml deleted file mode 100644 index 41258cc..0000000 --- a/blade-service/blade-openapi/src/main/resources/bootstrap.yml +++ /dev/null @@ -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 # 必须配置，负责自动刷新不生效 diff --git a/blade-service/blade-system/src/main/java/org/springblade/process/feign/BusinessProcessClient.java b/blade-service/blade-system/src/main/java/org/springblade/process/feign/BusinessProcessClient.java index e8ca2df..eb789f4 100644 --- a/blade-service/blade-system/src/main/java/org/springblade/process/feign/BusinessProcessClient.java +++ b/blade-service/blade-system/src/main/java/org/springblade/process/feign/BusinessProcessClient.java @@ -40,7 +40,7 @@ public class BusinessProcessClient implements IBusinessProcessClient { } @Override - public FR updateBusinessProcessStatus(@Validated @RequestBody BusinessProcessUpdateDTO param) { + public FR updateBusinessProcessStatus(@Validated @RequestBody BusinessProcessUpdateDTO param) { return FR.data(businessProcessService.updateBusinessProcessStatus(param)); } diff --git a/blade-service/blade-system/src/main/java/org/springblade/process/service/IBusinessProcessService.java b/blade-service/blade-system/src/main/java/org/springblade/process/service/IBusinessProcessService.java index 9dd744d..5e3e2ff 100644 --- a/blade-service/blade-system/src/main/java/org/springblade/process/service/IBusinessProcessService.java +++ b/blade-service/blade-system/src/main/java/org/springblade/process/service/IBusinessProcessService.java @@ -30,7 +30,7 @@ public interface IBusinessProcessService extends IService { * @param param * @return */ - boolean updateBusinessProcessStatus(BusinessProcessUpdateDTO param); + String updateBusinessProcessStatus(BusinessProcessUpdateDTO param); /** * 修改业务流程审批人 diff --git a/blade-service/blade-system/src/main/java/org/springblade/process/service/impl/BusinessProcessServiceImpl.java b/blade-service/blade-system/src/main/java/org/springblade/process/service/impl/BusinessProcessServiceImpl.java index 105651e..50984d3 100644 --- a/blade-service/blade-system/src/main/java/org/springblade/process/service/impl/BusinessProcessServiceImpl.java +++ b/blade-service/blade-system/src/main/java/org/springblade/process/service/impl/BusinessProcessServiceImpl.java @@ -116,7 +116,7 @@ public class BusinessProcessServiceImpl extends ServiceImpllambdaQuery() .eq(BusinessProcess::getProcessInstanceId, param.getProcessInstanceId()) @@ -128,9 +128,10 @@ public class BusinessProcessServiceImpl extends ServiceImpl