mirror of
http://git.pig4cloud.com/pig/pigx.git
synced 2026-11-11 05:19:00 +08:00
feat(pigx-flow-biz): 增加触发器异常终止处理，确保在事务外安全终止流程
- 在 TriggerServiceTask 中重构异常处理逻辑，将直接终止流程改为抛出特定异常 - 在 TaskServiceImpl 中增加触发器异常终止处理，确保在事务外安全终止流程 - 优化任务参数设置逻辑，提前获取 processInstanceId 确保终止流程可用 - 统一流程终止状态管理，使用 NodeStatusEnum.YZZ 表示已终止状态 - 调整代码格式，提高可读性和一致性 feat(pigx-flow-api): 扩展节点状态枚举 - 在 NodeStatusEnum 中新增 YZZ(9, "已终止") 状态，用于标识流程异常终止情况
This commit is contained in:
PIGCLOUD committed
1 parent
cd2ec46f82
commit
96361c21ec
3 files changed
+71
-21
No files matched your search
+9
-3
@@ -25,7 +25,7 @@ public enum NodeStatusEnum {
|
||||
* </p>
|
||||
*/
|
||||
WKS(0, "未开始"),
|
||||
|
||||
|
||||
/**
|
||||
* 进行中
|
||||
* <p>
|
||||
@@ -34,7 +34,7 @@ public enum NodeStatusEnum {
|
||||
* </p>
|
||||
*/
|
||||
JXZ(1, "进行中"),
|
||||
|
||||
|
||||
/**
|
||||
* 已结束
|
||||
* <p>
|
||||
@@ -42,7 +42,13 @@ public enum NodeStatusEnum {
|
||||
* 包括正常完成、跳过、驳回等所有结束状态。
|
||||
* </p>
|
||||
*/
|
||||
YJS(2, "已结束");
|
||||
YJS(2, "已结束"),
|
||||
|
||||
|
||||
/**
|
||||
* 已终止
|
||||
*/
|
||||
YZZ(9, "已终止");
|
||||
|
||||
/**
|
||||
* 状态编码
|
||||
|
||||
+59
-15
@@ -23,6 +23,7 @@ import com.pig4cloud.pigx.flow.support.utils.NodeUtil;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.flowable.common.engine.api.FlowableException;
|
||||
import org.flowable.engine.HistoryService;
|
||||
import org.flowable.engine.RuntimeService;
|
||||
import org.flowable.engine.TaskService;
|
||||
@@ -310,12 +311,14 @@ public class TaskServiceImpl implements ITaskService {
|
||||
taskService.setVariablesLocal(taskParamDto.getTaskId(), taskLocalParamMap);
|
||||
}
|
||||
|
||||
// 始终获取 processInstanceId，确保触发器终止时可用
|
||||
Task task = taskService.createTaskQuery().taskId(taskParamDto.getTaskId()).singleResult();
|
||||
taskParamDto.setProcessInstanceId(task.getProcessInstanceId());
|
||||
|
||||
if (CollUtil.isNotEmpty(taskParamDto.getFormData())) {
|
||||
// 获取当前任务
|
||||
Task task = taskService.createTaskQuery().taskId(taskParamDto.getTaskId()).singleResult();
|
||||
runtimeService.setVariables(task.getExecutionId(), taskParamDto.getFormData());
|
||||
|
||||
// 新增: 同步更新流程实例记录的主表单数据
|
||||
// 同步更新流程实例记录的主表单数据
|
||||
String processInstanceId = task.getProcessInstanceId();
|
||||
ProcessInstanceRecord processInstanceRecord = processInstanceRecordService.lambdaQuery()
|
||||
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
|
||||
@@ -349,11 +352,51 @@ public class TaskServiceImpl implements ITaskService {
|
||||
}
|
||||
}
|
||||
|
||||
taskService.complete(taskParamDto.getTaskId(), taskParamDto.getParamMap());
|
||||
try {
|
||||
taskService.complete(taskParamDto.getTaskId(), taskParamDto.getParamMap());
|
||||
} catch (FlowableException e) {
|
||||
String message = e.getMessage();
|
||||
if (StrUtil.isNotBlank(message) && message.startsWith("TRIGGER_TERMINATE:")) {
|
||||
return handleTriggerTerminate(message, taskParamDto.getProcessInstanceId());
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
|
||||
return R.ok();
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理触发器异常终止
|
||||
* <p>
|
||||
* 在 Flowable 事务回滚后执行，此时处于新的事务上下文中，
|
||||
* 可安全地删除流程实例和更新 ProcessInstanceRecord 状态。
|
||||
* </p>
|
||||
*
|
||||
* @param exceptionMessage 格式: TRIGGER_TERMINATE:{errorMessage}
|
||||
* @param processInstanceId 流程实例ID
|
||||
* @return R 失败结果
|
||||
*/
|
||||
private R handleTriggerTerminate(String exceptionMessage, String processInstanceId) {
|
||||
String errorMsg = StrUtil.removePrefix(exceptionMessage, "TRIGGER_TERMINATE:");
|
||||
|
||||
// 在 Flowable 事务外删除流程实例（新的命令上下文 = 新事务）
|
||||
try {
|
||||
runtimeService.deleteProcessInstance(processInstanceId, "触发器异常终止: " + errorMsg);
|
||||
} catch (Exception ex) {
|
||||
log.error("触发器终止-删除流程实例失败: processInstanceId={}, error={}", processInstanceId, ex.getMessage());
|
||||
}
|
||||
|
||||
// 更新 ProcessInstanceRecord 状态为已结束（与 stopProcessInstance 保持一致）
|
||||
processInstanceRecordService.lambdaUpdate()
|
||||
.set(ProcessInstanceRecord::getStatus, NodeStatusEnum.YJS.getCode())
|
||||
.set(ProcessInstanceRecord::getEndTime, new Date())
|
||||
.set(ProcessInstanceRecord::getFinishReason, String.valueOf(NodeStatusEnum.YZZ.getCode()))
|
||||
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
|
||||
.update();
|
||||
|
||||
return R.failed("触发器调用失败: " + errorMsg);
|
||||
}
|
||||
|
||||
/**
|
||||
* 前加签
|
||||
* <p>
|
||||
@@ -461,7 +504,7 @@ public class TaskServiceImpl implements ITaskService {
|
||||
// 停止流程实例后，更新流程实例记录状态
|
||||
processInstanceRecordService.lambdaUpdate()
|
||||
.set(ProcessInstanceRecord::getStatus, NodeStatusEnum.YJS.getCode())
|
||||
.set(ProcessInstanceRecord::getFinishReason, "9")
|
||||
.set(ProcessInstanceRecord::getFinishReason, String.valueOf(NodeStatusEnum.YZZ.getCode()))
|
||||
.set(ProcessInstanceRecord::getEndTime, new Date())
|
||||
.eq(ProcessInstanceRecord::getProcessInstanceId, reqProcessInstanceId)
|
||||
.update(new ProcessInstanceRecord());
|
||||
@@ -506,10 +549,10 @@ public class TaskServiceImpl implements ITaskService {
|
||||
|
||||
// 1. 获取任务并验证
|
||||
Task task = taskService.createTaskQuery()
|
||||
.taskId(taskParamDto.getTaskId())
|
||||
.taskTenantId(TenantContextHolder.getTenantId().toString())
|
||||
.taskAssignee(String.valueOf(userId))
|
||||
.singleResult();
|
||||
.taskId(taskParamDto.getTaskId())
|
||||
.taskTenantId(TenantContextHolder.getTenantId().toString())
|
||||
.taskAssignee(String.valueOf(userId))
|
||||
.singleResult();
|
||||
|
||||
if (task == null) {
|
||||
return R.failed("任务不存在或无权限操作");
|
||||
@@ -528,22 +571,23 @@ public class TaskServiceImpl implements ITaskService {
|
||||
|
||||
// 同步更新到 ProcessInstanceRecord
|
||||
ProcessInstanceRecord processInstanceRecord = processInstanceRecordService.lambdaQuery()
|
||||
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
|
||||
.one();
|
||||
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
|
||||
.one();
|
||||
|
||||
if (processInstanceRecord != null) {
|
||||
Map<String, Object> existingFormData = new HashMap<>();
|
||||
if (StrUtil.isNotBlank(processInstanceRecord.getFormData())) {
|
||||
existingFormData = objectMapper.readValue(processInstanceRecord.getFormData(),
|
||||
new TypeReference<Map<String, Object>>() {});
|
||||
new TypeReference<Map<String, Object>>() {
|
||||
});
|
||||
}
|
||||
existingFormData.putAll(taskParamDto.getFormData());
|
||||
String updatedFormData = objectMapper.writeValueAsString(existingFormData);
|
||||
|
||||
processInstanceRecordService.lambdaUpdate()
|
||||
.set(ProcessInstanceRecord::getFormData, updatedFormData)
|
||||
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
|
||||
.update();
|
||||
.set(ProcessInstanceRecord::getFormData, updatedFormData)
|
||||
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
|
||||
.update();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+3
-3
@@ -18,7 +18,6 @@ import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.flowable.common.engine.api.FlowableException;
|
||||
import org.flowable.engine.RuntimeService;
|
||||
import org.flowable.engine.delegate.DelegateExecution;
|
||||
import org.flowable.engine.delegate.JavaDelegate;
|
||||
import org.flowable.engine.impl.persistence.entity.ExecutionEntityImpl;
|
||||
@@ -193,8 +192,9 @@ public class TriggerServiceTask implements JavaDelegate {
|
||||
|
||||
private void handleError(String handler, DelegateExecution execution, String msg) {
|
||||
if ("TERMINATE".equals(handler)) {
|
||||
RuntimeService runtimeService = SpringUtil.getBean(RuntimeService.class);
|
||||
runtimeService.deleteProcessInstance(execution.getProcessInstanceId(), "触发器异常终止: " + msg);
|
||||
// 不在 JavaDelegate 内调用 deleteProcessInstance，因为后续异常会导致 Flowable 事务回滚
|
||||
// 抛出带特定前缀的异常，由上层 completeTask 捕获后在事务外执行终止操作
|
||||
throw new FlowableException("TRIGGER_TERMINATE:" + msg);
|
||||
}
|
||||
throw new FlowableException("触发器调用失败: " + msg);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user