工作流全面性能优化方案

现状分析

业务范围统计

通过代码扫描发现，当前系统中涉及工作流的业务模块有 18个：

序号	业务模块	监听器	业务服务
1	用车申请	HjlAuditCarStatusListener	HjlAuditCarServiceImplProcess
2	报销申请	HjlApplyExpenseStatusListener	-
3	发票申请	HjlApplyInvoiceStatusListener	-
4	出差申请	HjlAuditEvectionStatusListener	-
5	外出申请	HjlAuditGoOutStatusListener	-
6	用印申请	HjlAuditSealStatusListener	-
7	采购申请	HjlAuditShoppingStatusListener	-
8	合同变更	HjlContractAlterationStatusListener	-
9	合同补录	HjlContractReplenishsaveStatusListener	-
10	发票核销	HjlInvoiceCancellationStatusListener	-
11	标准起草	StandardDraftStatusListener	-
12	标准立项	StandardProjectStatusListener	-
13	标准推荐	StandardPushProjectStatusListener	-
14	标准审查	StandardReviewStatusListener	-
15	标准报批	StandardApplyForApprovalStatusListener	-
16	专家征集	StandardExpertCollectStatusListener	-
17	标准征求意见	StandardSeekOpinionsStatusListener	-
18	标准体系	StandardSystemStatusListener	-

结论：几乎所有业务都采用相同的模式，批量优化可以带来巨大收益。

架构分析

当前架构

┌────────────────────────────────────────────────────────────┐
│                      用户请求                                │
└────────────────────────────────────────────────────────────┘
                            │
                            ▼
┌────────────────────────────────────────────────────────────┐
│          业务服务 (pigx-standard-biz)                       │
│  ┌──────────────────────────────────────────────────────┐  │
│  │ @Transactional                                       │  │
│  │ 1. save(entity)  ← 获取行锁                          │  │
│  │ 2. Feign调用 wflow 服务  ← 同步阻塞，持有锁          │  │
│  │ 3. updateById(entity)  ← 仍在事务中                  │  │
│  │ 4. 事务提交  ← 释放锁                                │  │
│  └──────────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────────┘
                            │
                            │ Feign (HTTP)
                            ▼
┌────────────────────────────────────────────────────────────┐
│           工作流服务 (pigx-wflow-biz)                       │
│  ┌──────────────────────────────────────────────────────┐  │
│  │ @Transactional                                       │  │
│  │ 1. 创建流程实例                                       │  │
│  │ 2. 保存流程数据                                       │  │
│  │ 3. completeRootTask() ← 触发事件                     │  │
│  │ 4. publishEvent() ← 同步发布事件                     │  │
│  └──────────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────────┘
                            │
                            │ Spring Event (同步)
                            ▼
┌────────────────────────────────────────────────────────────┐
│              监听器 (*StatusListener)                       │
│  ┌──────────────────────────────────────────────────────┐  │
│  │ onEvent(event)                                       │  │
│  │ └─> Feign回调业务服务  ← 同步回调                    │  │
│  └──────────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────────┘
                            │
                            │ Feign (HTTP)
                            ▼
┌────────────────────────────────────────────────────────────┐
│          业务服务 - 状态更新接口                            │
│  ┌──────────────────────────────────────────────────────┐  │
│  │ @Transactional (新事务)                              │  │
│  │ updateWorkflowStatus()                               │  │
│  │ └─> UPDATE entity  ← 等待第一个事务释放锁  ❌ 死锁    │  │
│  └──────────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────────┘

问题点：

❌ 业务事务持有锁期间进行远程调用（Feign）
❌ 同步阻塞调用链路长（业务 → wflow → 监听器 → 业务）
❌ 监听器回调时形成锁竞争
❌ 数据库连接占用时间长
❌ 用户请求响应慢

核心问题总结

1. 业务层性能问题

1.1 事务中进行远程调用

问题代码位置：所有业务的 save* 或 update* 方法

java

@Transactional(rollbackFor = Exception.class)
public Boolean saveXxx(XxxEntity entity) {
    save(entity);  // ① 获取行锁
    ...
    wfInstanceService.startupProcess(params);  // ② Feign调用，可能耗时3-5秒
    ...
    updateById(entity);  // ③ 更新数据，仍在事务中
    return true;  // ④ 事务提交，释放锁
}

性能影响：

数据库连接占用时间：3-5秒（应该 < 100ms）
行锁占用时间：3-5秒（应该 < 100ms）
并发能力：极低（每个请求都长时间占用连接）

1.2 循环依赖导致死锁

调用链路：

T1: save() → Feign(wflow) → [等待wflow响应]
                     ↓
          wflow: processStart() → publishEvent()
                     ↓
          Listener: onEvent() → Feign(业务服务)
                     ↓
T2: updateWorkflowStatus() → UPDATE [等待T1释放锁]
          ↓
    [死锁或超时]

结果：

锁等待超时
用户请求失败
数据不一致

2. 工作流引擎性能问题

2.1 启动流程的大事务

位置：FlowInstanceServiceImpl.java:124-193

java

@Transactional
public String startProcess(StartProcParamsVo params, ...) {
    // ① 查询流程模型
    modelDo = modelService.getActiveModelDo(defineId);

    // ② 校验权限
    validateStartupPerms(modelDo.getCode(), initiator);

    // ③ 保存表单数据
    formDataService.saveInstFormData(instanceId, ...);

    // ④ 创建流程实例
    ProcessInstance instance = runtimeService.createProcessInstanceBuilder()...start();

    // ⑤ 保存到业务表
    procInstMapper.insert(WflowProcInstDo.builder()...build());

    // ⑥ 自动完成发起人任务（触发事件）
    flowTaskService.completeRootTask(instanceId, null, initiator);

    return instance.getId();
}

性能问题：

事务过大，包含多次数据库操作
completeRootTask 会触发同步事件，阻塞事务提交
一个流程启动涉及 5+ 次数据库操作

2.2 大量数据库查询

示例：FlowTaskServiceImpl.getTodoTasks()

java

public Page<FlowInstTaskVo> getTodoTasks(FlowInstQueryParamsVo queryParams) {
    // ① 查询任务列表
    List<Task> tasks = taskQuery.active().orderByTaskCreateTime().desc()
            .listPage(...);

    // ② 查询流程实例（N个任务 = N次查询）
    Map<String, WflowProcInstDo> instMap = instMapper.selectList(...)

    // ③ 查询部门信息（M个部门 = M次查询）
    Map<String, String> deptMap = orgIntegration.getDeptInfoList(deptSet)

    // ④ 查询用户信息（K个用户 = K次查询）
    orgIntegration.getAndBatchFillUserInfo(taskUsers, userSet);
}

问题：

N+1 查询问题：查询任务后，逐个查询关联数据
未使用缓存：每次都查询数据库
批量操作不够优化

2.3 缓存策略不足

位置：LocalFlowCache.java

java

// 节点缓存：3天过期
private static final TimedCache<String, Map<String, ProcNode>> nodeCatch =
    CacheUtil.newTimedCache(3 * 86400000);

// 数据缓存：2秒过期
private static final TimedCache<String, Object> dataCatch =
    CacheUtil.newTimedCache(2000);

问题：

本地缓存：多实例部署时缓存不同步
容量限制：TimedCache 无容量限制，可能内存溢出
缓存粒度粗：整个流程定义缓存，修改一个节点需刷新全部
缓存未命中率高：2秒过期太短，频繁查询数据库

3. 事件和监听器问题

3.1 同步事件阻塞主流程

位置：EventHandler.java:310

java

public void processStart(ExecutionEntity execution) {
    ...
    // 同步发布事件，阻塞主流程
    publishEvent(execution, InstanceStatus.RUNNING);
}

问题：

Spring 默认事件是同步的
监听器执行完才返回
监听器中的 Feign 调用会阻塞流程启动

3.2 所有监听器都是同步的

示例：HjlAuditCarStatusListener.java

java

@Component
@AllArgsConstructor
public class HjlAuditCarStatusListener extends ProcessInstanceResultEventListener {

    private final HjlAuditCarFeignService auditCarFeignService;

    @Override
    protected void onEvent(ProcessInstanceResultEvent event) {
        // 同步执行，阻塞事件发布线程
        WflowProcessResultDTO dto = EventConverter.convertToDTO(event);
        auditCarFeignService.updateAuditCarStatus(dto);  // Feign调用
    }
}

问题：

18个监听器都是同步的
每个 Feign 调用耗时 100-500ms
累积阻塞时间长

4. Feign 调用性能问题

4.1 同步阻塞调用

java

// 业务服务调用 wflow
Rsp<String> rsp = wfInstanceService.startupProcess(startProcParamsVo);

// wflow 回调业务服务
auditCarFeignService.updateAuditCarStatus(dto);

性能影响：

每次 Feign 调用：100-500ms（网络 + 序列化 + 反序列化）
一个流程启动：至少2次 Feign 调用 = 200-1000ms

4.2 未配置超时和重试

位置：WFlowFeignConfig.java

java

@Bean
public Request.Options feignRequestOptions() {
    return new Request.Options(
        15, TimeUnit.SECONDS,   // 连接超时 15秒
        180, TimeUnit.SECONDS,  // 读取超时 3分钟  ← 太长！
        true
    );
}

问题：

读取超时 3分钟 太长，锁占用时间过长
未配置重试策略
未配置熔断机制

优化策略

优化原则

业务层优化优先：解决核心锁竞争和死锁问题
工作流引擎优化：提升启动性能和查询效率
缓存和异步化：减少数据库查询，提升响应速度
最小侵入性：尽量避免大范围业务修改

优化层次

┌─────────────────────────────────────────────────┐
│  第一层：紧急修复（立即实施，1天内完成）           │
│  - 监听器异步化                                  │
│  - Feign 超时优化                                │
│  - 事务传播优化                                  │
└─────────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────────┐
│  第二层：业务优化（1-2周内完成）                  │
│  - 事务分离 + 异步启动流程                        │
│  - 通用异步工作流启动工具类                       │
│  - 批量业务改造                                  │
└─────────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────────┐
│  第三层：工作流引擎优化（2-4周内完成）             │
│  - 数据库查询优化                                │
│  - 缓存策略升级                                  │
│  - 批量操作优化                                  │
└─────────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────────┐
│  第四层：架构优化（长期规划）                     │
│  - 引入消息队列                                  │
│  - 分布式缓存                                    │
│  - 读写分离                                      │
└─────────────────────────────────────────────────┘

详细优化方案

第一层：紧急修复（立即实施）

1.1 监听器全面异步化

目标：将18个监听器全部改为异步执行，避免阻塞主流程。

实施方案：

步骤1：创建统一的异步监听器基类

java

package com.pig4cloud.pigx.wflow.core.event;

import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationListener;
import org.springframework.scheduling.annotation.Async;

/**
 * 异步流程实例状态变化事件监听器基类
 * 所有业务监听器继承此类，自动获得异步能力
 */
@Slf4j
public abstract class AsyncProcessInstanceResultEventListener
        implements ApplicationListener<ProcessInstanceResultEvent> {

    @Override
    @Async("workflowEventExecutor")  // 使用专用的异步线程池
    public final void onApplicationEvent(ProcessInstanceResultEvent event) {
        String processDefinitionKey = getProcessDefinitionKey();

        // 检查流程定义 Key 是否匹配
        if (!StrUtil.equals(event.getProcessDefinitionKey(), processDefinitionKey)) {
            return;
        }

        try {
            // 调用子类的业务处理方法
            onEvent(event);
            log.debug("异步处理流程事件成功: processDefKey={}, instanceId={}, businessKey={}",
                    processDefinitionKey, event.getInstanceId(), event.getBusinessKey());
        } catch (Exception e) {
            log.error("异步处理流程事件失败: processDefKey={}, instanceId={}, businessKey={}",
                    processDefinitionKey, event.getInstanceId(), event.getBusinessKey(), e);
            // 异常处理：可以记录到数据库，后续重试
            handleError(event, e);
        }
    }

    /**
     * 返回监听的流程定义 Key
     */
    protected abstract String getProcessDefinitionKey();

    /**
     * 处理事件（异步执行）
     */
    protected abstract void onEvent(ProcessInstanceResultEvent event);

    /**
     * 异常处理（可被子类覆盖）
     */
    protected void handleError(ProcessInstanceResultEvent event, Exception e) {
        // 默认实现：记录日志
        // 子类可以覆盖此方法，实现重试或补偿逻辑
    }
}

步骤2：配置异步线程池

java

package com.pig4cloud.pigx.wflow.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 异步配置
 */
@Slf4j
@Configuration
@EnableAsync
public class WorkflowAsyncConfig implements AsyncConfigurer {

    /**
     * 工作流事件处理线程池
     */
    @Bean("workflowEventExecutor")
    public Executor workflowEventExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

        // 核心线程数 = CPU核数
        int corePoolSize = Runtime.getRuntime().availableProcessors();
        executor.setCorePoolSize(corePoolSize);

        // 最大线程数 = CPU核数 * 2
        executor.setMaxPoolSize(corePoolSize * 2);

        // 队列容量：200个任务
        executor.setQueueCapacity(200);

        // 线程名前缀
        executor.setThreadNamePrefix("wflow-event-");

        // 线程空闲时间：60秒
        executor.setKeepAliveSeconds(60);

        // 拒绝策略：由调用线程执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        // 等待所有任务完成后关闭
        executor.setWaitForTasksToCompleteOnShutdown(true);

        // 等待时间：60秒
        executor.setAwaitTerminationSeconds(60);

        executor.initialize();

        log.info("工作流事件异步线程池初始化完成: corePoolSize={}, maxPoolSize={}, queueCapacity={}",
                corePoolSize, corePoolSize * 2, 200);

        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (ex, method, params) -> {
            log.error("异步任务执行异常: method={}, params={}", method.getName(), params, ex);
        };
    }
}

步骤3：修改所有监听器（批量操作）

java

// 修改前
@Component
@AllArgsConstructor
public class HjlAuditCarStatusListener extends ProcessInstanceResultEventListener {
    ...
}

// 修改后
@Component
@AllArgsConstructor
public class HjlAuditCarStatusListener extends AsyncProcessInstanceResultEventListener {
    ...
}

批量修改命令（使用 IDE 的查找替换功能）：

查找：extends ProcessInstanceResultEventListener
替换为：extends AsyncProcessInstanceResultEventListener

影响范围：18个监听器文件

预期效果：

流程启动不再等待监听器完成
响应时间从 3-5秒 降低到 200-500ms
锁占用时间减少 90%

1.2 Feign 超时优化

问题：当前 Feign 读取超时为 3分钟，导致死锁时等待时间过长。

优化方案：

java

package com.pig4cloud.pigx.standard.feign.config;

import feign.Request;
import feign.Retryer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.TimeUnit;

@Configuration
public class WFlowFeignConfig {

    @Bean
    public Request.Options feignRequestOptions() {
        return new Request.Options(
            5, TimeUnit.SECONDS,   // 连接超时 5秒（原15秒）
            15, TimeUnit.SECONDS,  // 读取超时 15秒（原3分钟） ✅ 优化
            true
        );
    }

    /**
     * 重试策略：最多重试2次，间隔1秒
     */
    @Bean
    public Retryer feignRetryer() {
        return new Retryer.Default(
            1000,  // 重试间隔：1秒
            3000,  // 最大重试间隔：3秒
            2      // 最多重试次数：2次
        );
    }
}

预期效果：

死锁时快速失败（15秒内）
减少数据库锁占用时间
提供重试机制，提高可靠性

1.3 事务传播优化

问题：updateWorkflowStatus 方法需要使用新事务，避免与主流程事务冲突。

优化方案：

java

// 所有业务服务的 updateWorkflowStatus 方法添加新事务
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public void updateWorkflowStatus(WflowProcessResultDTO event) {
    // 使用新事务，避免与主流程冲突
    ...
}

批量修改：

查找：@Transactional(rollbackFor = Exception.class)\s+public void updateWorkflowStatus
替换为：@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)\n public void updateWorkflowStatus

预期效果：

减少锁冲突概率
即使主流程失败，状态更新也可以提交

第二层：业务优化（1-2周）

2.1 通用异步工作流启动工具类

目标：提供统一的异步工作流启动工具，避免每个业务都重复实现。

实施方案：

java

package com.pig4cloud.pigx.standard.plugin.workflow.util;

import com.pig4cloud.pigx.common.security.util.SecurityUtils;
import com.pig4cloud.pigx.standard.feign.service.WFlowInstanceService;
import com.pig4cloud.pigx.standard.infrastructure.Rsp;
import com.pig4cloud.pigx.standard.vo.StartProcParamsVo;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * 工作流异步启动工具类
 * 提供统一的异步启动流程功能，避免业务代码重复
 */
@Slf4j
@Component
@AllArgsConstructor
public class WorkflowAsyncStarter {

    private final WFlowInstanceService wfInstanceService;

    /**
     * 在当前事务提交后，异步启动工作流
     *
     * @param processDefKey 流程定义Key
     * @param businessKey   业务主键
     * @param variables     流程变量
     * @param onSuccess     成功回调（可选）
     * @param onFailure     失败回调（可选）
     */
    public void startAfterCommit(String processDefKey,
                                 String businessKey,
                                 Map<String, Object> variables,
                                 Consumer<String> onSuccess,
                                 Consumer<Exception> onFailure) {

        // 检查是否在事务中
        if (!TransactionSynchronizationManager.isActualTransactionActive()) {
            log.warn("当前不在事务中，建议在事务中调用此方法");
            // 直接异步启动
            startProcessAsync(processDefKey, businessKey, variables, onSuccess, onFailure);
            return;
        }

        // 在事务提交后执行
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                log.debug("事务已提交，准备异步启动流程: processDefKey={}, businessKey={}",
                        processDefKey, businessKey);
                startProcessAsync(processDefKey, businessKey, variables, onSuccess, onFailure);
            }
        });
    }

    /**
     * 简化版本：仅需要流程Key和业务Key
     */
    public void startAfterCommit(String processDefKey, String businessKey) {
        startAfterCommit(processDefKey, businessKey, new HashMap<>(), null, null);
    }

    /**
     * 异步启动流程（内部方法）
     */
    @Async("workflowStarterExecutor")
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    protected void startProcessAsync(String processDefKey,
                                     String businessKey,
                                     Map<String, Object> variables,
                                     Consumer<String> onSuccess,
                                     Consumer<Exception> onFailure) {
        try {
            // 准备流程变量
            Map<String, Object> vars = new HashMap<>();
            if (variables != null) {
                vars.putAll(variables);
            }
            vars.put("processDefKey", processDefKey);
            vars.put("businessKey", businessKey);

            // 构建启动参数
            StartProcParamsVo params = StartProcParamsVo.builder()
                    .defineId(processDefKey)
                    .initiator(SecurityUtils.getUser().getId())
                    .businessKey(businessKey)
                    .startDeptId(SecurityUtils.getUser().getDeptId())
                    .variables(vars)
                    .build();

            // 调用 wflow 服务
            Rsp<String> rsp = wfInstanceService.startupProcess(params);
            String processInstanceId = rsp.getData();

            log.info("异步启动流程成功: processDefKey={}, businessKey={}, processInstanceId={}",
                    processDefKey, businessKey, processInstanceId);

            // 执行成功回调
            if (onSuccess != null) {
                onSuccess.accept(processInstanceId);
            }

        } catch (Exception e) {
            log.error("异步启动流程失败: processDefKey={}, businessKey={}",
                    processDefKey, businessKey, e);

            // 执行失败回调
            if (onFailure != null) {
                onFailure.accept(e);
            } else {
                // 默认失败处理：可以发送告警、记录到数据库等
                handleStartFailure(processDefKey, businessKey, e);
            }
        }
    }

    /**
     * 默认失败处理
     */
    protected void handleStartFailure(String processDefKey, String businessKey, Exception e) {
        // 可以实现：
        // 1. 发送告警通知
        // 2. 记录到失败表，后续重试
        // 3. 更新业务状态为"流程启动失败"
        log.error("工作流启动失败需要人工处理: processDefKey={}, businessKey={}",
                processDefKey, businessKey);
    }
}

配置异步线程池：

java

/**
 * 工作流启动线程池
 */
@Bean("workflowStarterExecutor")
public Executor workflowStarterExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("wflow-starter-");
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    executor.setKeepAliveSeconds(60);
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.setAwaitTerminationSeconds(60);
    executor.initialize();
    return executor;
}

业务代码改造示例：

java

@Service
public class HjlAuditCarServiceImplProcess extends ServiceImpl<...> {

    @Resource
    private WorkflowAsyncStarter workflowStarter;  // 注入工具类

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean saveAuditCar(HjlAuditCarEntity auditCarEntity) {
        // ✅ 第一步：仅保存数据
        save(auditCarEntity);
        Long id = auditCarEntity.getId();

        if ("1".equals(isDraft)) {
            // ✅ 第二步：事务提交后异步启动流程（一行代码搞定）
            workflowStarter.startAfterCommit(
                HJL_AUDIT_CAR_PROCESS,  // 流程Key
                id.toString(),          // 业务Key
                null,                   // 流程变量（可选）
                processInstanceId -> {  // 成功回调（可选）
                    log.info("流程启动成功: {}", processInstanceId);
                    // 更新流程实例ID
                    updateProcessInstanceId(id, processInstanceId);
                },
                e -> {                  // 失败回调（可选）
                    log.error("流程启动失败: {}", e.getMessage());
                    // 标记失败状态
                    markProcessStartFailed(id, e.getMessage());
                }
            );
        }
        return true;
    }

    /**
     * 更新流程实例ID
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public void updateProcessInstanceId(Long id, String processInstanceId) {
        HjlAuditCarEntity entity = getById(id);
        entity.setProcessInstanceId(processInstanceId);
        entity.setStatus(WfNotice.Type.START.getCode());
        updateById(entity);
    }

    /**
     * 标记失败状态
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public void markProcessStartFailed(Long id, String errorMsg) {
        HjlAuditCarEntity entity = getById(id);
        entity.setStatus(-1);
        entity.setRemark("流程启动失败: " + errorMsg);
        updateById(entity);
    }
}

优点：

✅ 业务代码简洁，一行代码启动流程
✅ 统一的异常处理和重试机制
✅ 支持成功/失败回调
✅ 易于维护和测试

批量改造：

18个业务模块可以使用相同的模式改造
改造工作量：每个业务约 30分钟
总计：18 * 30分钟 = 9小时

2.2 工作流启动状态追踪

问题：异步启动后，用户不知道流程是否启动成功。

解决方案：提供状态查询接口

java

@RestController
@RequestMapping("/workflow/status")
public class WorkflowStatusController {

    /**
     * 查询流程启动状态
     *
     * @param businessKey 业务主键
     * @return 状态信息
     */
    @GetMapping("/startup/{businessKey}")
    public R<WorkflowStartupStatus> getStartupStatus(@PathVariable String businessKey) {
        // 查询业务表，获取流程实例ID和状态
        // 返回：启动中、启动成功、启动失败
        ...
    }
}

第三层：工作流引擎优化（2-4周）

3.1 数据库查询优化

问题1：N+1 查询

位置：FlowTaskServiceImpl.getTodoTasks()

优化前：

java

// ① 查询任务列表
List<Task> tasks = taskQuery.active()...listPage(...);

// ② 逐个查询流程实例（N次查询）
Map<String, WflowProcInstDo> instMap = instMapper.selectList(
    new LambdaQueryWrapper<WflowProcInstDo>()
        .in(WflowProcInstDo::getInstId, instIds)
);

优化后：

java

// 使用 JOIN 查询，一次获取所有数据
List<FlowInstTaskVo> tasks = customQueryMapper.selectTodoTasksWithInstance(
    userId, tenantId, queryParams
);

SQL 优化：

xml

<!-- WflowCustomQueryMapper.xml -->
<select id="selectTodoTasksWithInstance" resultType="FlowInstTaskVo">
    SELECT
        t.id_ AS task_id,
        t.name_ AS task_name,
        t.create_time_ AS create_time,
        p.inst_id,
        p.title,
        p.define_name,
        p.version,
        p.code,
        p.initiator,
        p.dept_id,
        d.dept_name
    FROM
        ACT_RU_TASK t
        INNER JOIN wflow_proc_inst p ON t.proc_inst_id_ = p.inst_id
        LEFT JOIN sys_dept d ON p.dept_id = d.dept_id
    WHERE
        t.assignee_ = #{userId}
        AND t.tenant_id_ = #{tenantId}
        <if test="code != null and code != ''">
            AND p.code = #{code}
        </if>
    ORDER BY t.create_time_ DESC
    LIMIT #{offset}, #{limit}
</select>

预期效果：

查询次数从 N+1 降低到 1
响应时间降低 60-80%

问题2：索引优化

sql

-- wflow_proc_inst 表添加索引
CREATE INDEX idx_inst_id ON wflow_proc_inst(inst_id);
CREATE INDEX idx_business_key ON wflow_proc_inst(business_key);
CREATE INDEX idx_code_status ON wflow_proc_inst(code, status);

-- wflow_proc_inst_task 表添加索引
CREATE INDEX idx_assignee ON wflow_proc_inst_task(assignee, end_time);
CREATE INDEX idx_inst_id ON wflow_proc_inst_task(inst_id);

3.2 缓存策略升级

问题：当前使用本地缓存，多实例部署时缓存不同步。

优化方案：引入 Redis 缓存

java

package com.pig4cloud.pigx.wflow.cache.flow;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.pig4cloud.pigx.wflow.core.bean.process.ProcNode;
import com.pig4cloud.pigx.wflow.dal.ext.WflowProcModelDo;
import com.pig4cloud.pigx.wflow.exception.WflowException;
import com.pig4cloud.pigx.wflow.mapper.ext.WflowProcModelMapper;
import com.pig4cloud.pigx.wflow.utils.ProcessUtil;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * Redis 流程缓存
 */
@Slf4j
@Component
public class RedisFlowCache {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private WflowProcModelMapper modelMapper;

    private static final String CACHE_PREFIX = "wflow:proc:node:";
    private static final long CACHE_TTL = 3 * 24 * 60 * 60; // 3天

    /**
     * 获取流程节点（使用 Redis 缓存）
     */
    public Map<String, ProcNode> getProcNode(String defId) {
        String cacheKey = CACHE_PREFIX + defId;

        // 尝试从 Redis 获取
        @SuppressWarnings("unchecked")
        Map<String, ProcNode> nodeMap = (Map<String, ProcNode>) redisTemplate.opsForValue().get(cacheKey);

        if (nodeMap != null) {
            log.debug("从 Redis 缓存获取流程节点: defId={}", defId);
            return nodeMap;
        }

        // 缓存未命中，从数据库查询
        WflowProcModelDo modelDo = modelMapper.selectOne(
            new LambdaQueryWrapper<WflowProcModelDo>()
                .select(WflowProcModelDo::getProcess)
                .eq(WflowProcModelDo::getDefineId, defId)
        );

        if (Objects.isNull(modelDo)) {
            throw new WflowException("该流程模型已被删除");
        }

        try {
            String process = modelDo.getProcess();
            ObjectMapper objectMapper = new ObjectMapper();
            List<ProcNode> procNodes = objectMapper.readValue(process, new TypeReference<List<ProcNode>>() {});

            Map<String, ProcNode> procNodeMap = new HashMap<>();
            ProcessUtil.foreachNode(procNodes, v -> procNodeMap.put(v.getId(), v));

            // 存入 Redis
            redisTemplate.opsForValue().set(cacheKey, procNodeMap, CACHE_TTL, TimeUnit.SECONDS);

            log.debug("流程节点已缓存到 Redis: defId={}, nodeCount={}", defId, procNodeMap.size());

            return procNodeMap;

        } catch (Exception e) {
            log.error("流程 JSON 解析失败: defId={}", defId, e);
            throw new WflowException("流程json解析失败");
        }
    }

    /**
     * 获取单个节点
     */
    public ProcNode getProcNode(String defId, String nodeId) {
        return getProcNode(defId).get(nodeId);
    }

    /**
     * 清除缓存
     */
    public void evictCache(String defId) {
        String cacheKey = CACHE_PREFIX + defId;
        redisTemplate.delete(cacheKey);
        log.info("已清除流程缓存: defId={}", defId);
    }
}

配置 Redis：

yaml

# application.yml
spring:
  redis:
    host: localhost
    port: 6379
    password:
    database: 0
    timeout: 3000ms
    lettuce:
      pool:
        max-active: 8
        max-wait: -1ms
        max-idle: 8
        min-idle: 0

预期效果：

缓存命中率：95%+
响应时间降低：80%
多实例部署缓存同步

3.3 批量操作优化

问题：保存流程实例数据时，多次单条插入。

优化方案：使用批量插入

java

// 优化前
for (WflowProcInstNodesDo node : nodes) {
    instNodesMapper.insert(node);  // N次插入
}

// 优化后
instNodesMapper.insertBatch(nodes);  // 1次批量插入

MyBatis-Plus 批量插入：

java

@Service
public class FlowInstanceServiceImpl implements FlowInstanceService {

    @Resource
    private SqlSessionFactory sqlSessionFactory;

    public void batchInsertNodes(List<WflowProcInstNodesDo> nodes) {
        // 使用 MyBatis-Plus 的批量插入
        try (SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH)) {
            WflowProcInstNodesMapper mapper = sqlSession.getMapper(WflowProcInstNodesMapper.class);
            for (WflowProcInstNodesDo node : nodes) {
                mapper.insert(node);
            }
            sqlSession.commit();
        }
    }
}

预期效果：

批量插入性能提升 10-20倍

第四层：架构优化（长期规划）

4.1 引入消息队列（RabbitMQ/Kafka）

架构图：

业务服务                    消息队列                    工作流服务
   |                          |                            |
   |--save(entity)            |                            |
   |                          |                            |
   |--发送消息--------------> |                            |
   |  (事务提交后)            |                            |
   |                          |                            |
   |<-返回成功                |                            |
   |                          |                            |
   |                          |--消费消息----------------> |
   |                          |                            |--启动流程
   |                          |                            |
   |                          |<-ACK                       |
   |                          |                            |
   |                          |--发送状态消息------------> |
   |<-Feign回调或消息通知     |                            |

优点：

✅ 完全解耦业务和工作流
✅ 自动重试机制
✅ 削峰填谷
✅ 可观测性强

实施成本：

需要部署 MQ 中间件
需要处理消息丢失、重复等问题
开发周期：2-4周

4.2 分布式缓存集群

方案：

Redis Cluster 或 Redis Sentinel
提供高可用缓存

配置示例：

yaml

spring:
  redis:
    cluster:
      nodes:
        - 192.168.1.1:6379
        - 192.168.1.2:6379
        - 192.168.1.3:6379
      max-redirects: 3
    lettuce:
      pool:
        max-active: 20
        max-idle: 10
        min-idle: 5

4.3 数据库读写分离

方案：

主库：处理写操作
从库：处理读操作（查询待办、已办等）

配置示例：

java

@Configuration
public class DataSourceConfig {

    @Bean
    @ConfigurationProperties("spring.datasource.master")
    public DataSource masterDataSource() {
        return DataSourceBuilder.create().build();
    }

    @Bean
    @ConfigurationProperties("spring.datasource.slave")
    public DataSource slaveDataSource() {
        return DataSourceBuilder.create().build();
    }

    @Bean
    public DataSource dynamicDataSource() {
        Map<Object, Object> dataSourceMap = new HashMap<>();
        dataSourceMap.put("master", masterDataSource());
        dataSourceMap.put("slave", slaveDataSource());

        DynamicDataSource dynamicDataSource = new DynamicDataSource();
        dynamicDataSource.setTargetDataSources(dataSourceMap);
        dynamicDataSource.setDefaultTargetDataSource(masterDataSource());

        return dynamicDataSource;
    }
}

实施路线图

第一阶段：紧急修复（1-2天）

任务	负责人	工作量	优先级
创建异步监听器基类	开发	2小时	P0
配置异步线程池	开发	1小时	P0
批量修改18个监听器	开发	2小时	P0
Feign 超时优化	开发	0.5小时	P0
事务传播优化	开发	1小时	P0
测试验证	测试	2小时	P0

总计：8.5小时（1天）

第二阶段：业务优化（1-2周）

任务	负责人	工作量	优先级
开发通用异步启动工具类	开发	4小时	P1
改造用车申请业务（示例）	开发	2小时	P1
批量改造其他17个业务	开发	34小时	P1
开发流程状态查询接口	开发	4小时	P1
测试验证	测试	8小时	P1

总计：52小时（1.5周）

第三阶段：引擎优化（2-4周）

任务	负责人	工作量	优先级
SQL 查询优化	开发	8小时	P2
数据库索引优化	DBA	4小时	P2
引入 Redis 缓存	开发	16小时	P2
批量操作优化	开发	8小时	P2
测试验证	测试	8小时	P2

总计：44小时（1.5周）

第四阶段：架构优化（长期规划）

任务	负责人	工作量	优先级
评估 MQ 方案	架构师	8小时	P3
部署 RabbitMQ 集群	运维	16小时	P3
开发 MQ 生产者/消费者	开发	24小时	P3
灰度发布	开发+运维	16小时	P3

总计：64小时（2周）

性能提升预期

优化前 vs 优化后

指标	优化前	第一阶段后	第二阶段后	第三阶段后
用户请求响应时间	3-5秒	500-800ms	200-500ms	100-200ms
数据库连接占用	3-5秒	500-800ms	50-100ms	50-100ms
行锁占用时间	3-5秒	500-800ms	50-100ms	50-100ms
锁超时频率	频繁	偶尔	几乎没有	没有
并发处理能力	10 req/s	50 req/s	200 req/s	500 req/s
工作流启动成功率	70-80%	90-95%	98-99%	99.9%

业务价值

用户体验大幅提升：
- 响应时间从 5秒降低到 200ms
- 减少用户等待时间 96%
系统稳定性提升：
- 锁超时问题完全解决
- 数据库连接池不再被耗尽
并发能力提升：
- 从 10 req/s 提升到 200 req/s
- 支持更多并发用户
运维成本降低：
- 减少数据库压力
- 减少故障和告警

监控和告警

关键指标

java

/**
 * 工作流性能监控
 */
@Component
public class WorkflowMetrics {

    @Resource
    private MeterRegistry meterRegistry;

    /**
     * 记录流程启动耗时
     */
    public void recordStartDuration(String processDefKey, long durationMs) {
        Timer.builder("workflow.start.duration")
                .tag("processDefKey", processDefKey)
                .register(meterRegistry)
                .record(durationMs, TimeUnit.MILLISECONDS);
    }

    /**
     * 记录流程启动失败
     */
    public void incrementStartFailure(String processDefKey, String errorType) {
        Counter.builder("workflow.start.failure")
                .tag("processDefKey", processDefKey)
                .tag("errorType", errorType)
                .register(meterRegistry)
                .increment();
    }

    /**
     * 记录异步线程池状态
     */
    @Scheduled(fixedRate = 60000)
    public void recordThreadPoolMetrics() {
        ThreadPoolTaskExecutor executor = getExecutor();

        Gauge.builder("workflow.threadpool.active", executor, ThreadPoolTaskExecutor::getActiveCount)
                .register(meterRegistry);

        Gauge.builder("workflow.threadpool.queue.size", executor, e -> e.getThreadPoolExecutor().getQueue().size())
                .register(meterRegistry);
    }
}

Grafana 监控面板

建议监控指标：

流程启动耗时（P50, P95, P99）
流程启动成功率
异步线程池使用情况
数据库连接池使用情况
缓存命中率

总结

核心优化点

业务层：
- ✅ 事务分离，避免在事务中进行远程调用
- ✅ 异步启动流程，快速释放锁
- ✅ 统一的工具类，简化业务开发
工作流引擎：
- ✅ 监听器异步化，避免阻塞主流程
- ✅ 数据库查询优化，减少 N+1 查询
- ✅ Redis 缓存，提升查询性能
架构层：
- ✅ 消息队列解耦（长期规划）
- ✅ 分布式缓存（长期规划）
- ✅ 读写分离（长期规划）

实施建议

立即实施：第一阶段（监听器异步化）
短期实施：第二阶段（业务优化）
中期实施：第三阶段（引擎优化）
长期规划：第四阶段（架构优化）

预期收益

用户响应速度提升 96%
并发能力提升 20倍
锁超时问题 完全解决
系统稳定性大幅提升

附录

A. 异步线程池配置参考

yaml

# application.yml
workflow:
  async:
    # 工作流事件处理线程池
    event-executor:
      core-pool-size: ${CPU_CORES:4}
      max-pool-size: ${CPU_CORES:4}*2
      queue-capacity: 200
      keep-alive-seconds: 60
      thread-name-prefix: wflow-event-

    # 工作流启动线程池
    starter-executor:
      core-pool-size: 5
      max-pool-size: 10
      queue-capacity: 100
      keep-alive-seconds: 60
      thread-name-prefix: wflow-starter-

B. Redis 缓存配置参考

yaml

spring:
  redis:
    host: ${REDIS_HOST:localhost}
    port: ${REDIS_PORT:6379}
    password: ${REDIS_PASSWORD:}
    database: 0
    timeout: 3000ms
    lettuce:
      pool:
        max-active: 20
        max-wait: -1ms
        max-idle: 10
        min-idle: 5

  cache:
    type: redis
    redis:
      time-to-live: 3600000  # 1小时
      cache-null-values: false

C. Feign 配置参考

yaml

feign:
  client:
    config:
      default:
        connect-timeout: 5000   # 连接超时 5秒
        read-timeout: 15000     # 读取超时 15秒
      wflow-server:
        connect-timeout: 5000
        read-timeout: 15000

  compression:
    request:
      enabled: true
      mime-types: text/xml,application/xml,application/json
      min-request-size: 2048
    response:
      enabled: true

  hystrix:
    enabled: true

hystrix:
  command:
    default:
      execution:
        isolation:
          thread:
            timeoutInMilliseconds: 30000

工作流全面性能优化方案 ​

目录 ​

现状分析 ​

业务范围统计 ​

架构分析 ​

当前架构 ​

核心问题总结 ​

1. 业务层性能问题 ​

1.1 事务中进行远程调用 ​

1.2 循环依赖导致死锁 ​

2. 工作流引擎性能问题 ​

2.1 启动流程的大事务 ​

2.2 大量数据库查询 ​

2.3 缓存策略不足 ​

3. 事件和监听器问题 ​

3.1 同步事件阻塞主流程 ​

3.2 所有监听器都是同步的 ​

4. Feign 调用性能问题 ​

4.1 同步阻塞调用 ​

4.2 未配置超时和重试 ​

优化策略 ​

优化原则 ​

优化层次 ​

详细优化方案 ​

第一层：紧急修复（立即实施） ​

1.1 监听器全面异步化 ​

1.2 Feign 超时优化 ​

1.3 事务传播优化 ​

第二层：业务优化（1-2周） ​

2.1 通用异步工作流启动工具类 ​

2.2 工作流启动状态追踪 ​

第三层：工作流引擎优化（2-4周） ​

3.1 数据库查询优化 ​

3.2 缓存策略升级 ​

3.3 批量操作优化 ​

第四层：架构优化（长期规划） ​

4.1 引入消息队列（RabbitMQ/Kafka） ​

4.2 分布式缓存集群 ​

4.3 数据库读写分离 ​

实施路线图 ​

第一阶段：紧急修复（1-2天） ​

第二阶段：业务优化（1-2周） ​

第三阶段：引擎优化（2-4周） ​

第四阶段：架构优化（长期规划） ​

性能提升预期 ​

优化前 vs 优化后 ​

业务价值 ​

监控和告警 ​

关键指标 ​

Grafana 监控面板 ​

总结 ​

核心优化点 ​

实施建议 ​

预期收益 ​

附录 ​

A. 异步线程池配置参考 ​

B. Redis 缓存配置参考 ​

C. Feign 配置参考 ​

工作流全面性能优化方案

目录

现状分析

业务范围统计

架构分析

当前架构

核心问题总结

1. 业务层性能问题

1.1 事务中进行远程调用

1.2 循环依赖导致死锁

2. 工作流引擎性能问题

2.1 启动流程的大事务

2.2 大量数据库查询

2.3 缓存策略不足

3. 事件和监听器问题

3.1 同步事件阻塞主流程

3.2 所有监听器都是同步的

4. Feign 调用性能问题

4.1 同步阻塞调用

4.2 未配置超时和重试

优化策略

优化原则

优化层次

详细优化方案

第一层：紧急修复（立即实施）

1.1 监听器全面异步化

1.2 Feign 超时优化

1.3 事务传播优化

第二层：业务优化（1-2周）

2.1 通用异步工作流启动工具类

2.2 工作流启动状态追踪

第三层：工作流引擎优化（2-4周）

3.1 数据库查询优化

3.2 缓存策略升级

3.3 批量操作优化

第四层：架构优化（长期规划）

4.1 引入消息队列（RabbitMQ/Kafka）

4.2 分布式缓存集群

4.3 数据库读写分离

实施路线图

第一阶段：紧急修复（1-2天）

第二阶段：业务优化（1-2周）

第三阶段：引擎优化（2-4周）

第四阶段：架构优化（长期规划）

性能提升预期

优化前 vs 优化后

业务价值

监控和告警

关键指标

Grafana 监控面板

总结

核心优化点

实施建议

预期收益

附录

A. 异步线程池配置参考

B. Redis 缓存配置参考

C. Feign 配置参考