引言：工作流引擎的重要性

公司需要开发一个请假系统，业务逻辑复杂：普通员工请假需要直属领导审批，超过3天需要部门领导审批，超过7天需要总经理审批？或者报销流程更复杂：填写报销单→部门领导审批→财务审核→总经理审批→出纳付款？再或者每次业务流程变更，都需要重新开发部署？

这就是传统硬编码工作流的典型痛点。今天我们就来聊聊如何用SpringBoot + Flowable构建一个可视化的工作流引擎，通过自定义节点支持各种复杂的业务场景。

为什么传统工作流实现方式效率低？

先说说为什么传统的硬编码方式效率低下。

想象一下，你是一家企业的后端工程师。公司有10种业务流程：

• 请假流程
• 报销流程
• 采购流程
• 入职流程
• 离职流程
• 等等...

如果按传统方式开发：

1. 每个流程都要写一套状态机逻辑
2. 流程变更需要重新开发部署
3. 代码重复率高，维护成本大
4. 业务人员无法自主配置流程

这会导致什么问题？

• 开发效率低：大量重复工作
• 维护成本高：代码分散难管理
• 响应速度慢：流程变更周期长
• 灵活性差：无法快速适应业务变化

技术选型：为什么选择这些技术？

Flowable：企业级工作流引擎

Flowable是业界领先的工作流引擎：

• 标准化：基于BPMN 2.0标准
• 功能完整：支持复杂业务流程
• 性能优异：企业级性能表现
• 社区活跃：持续更新维护

自定义节点：业务逻辑的灵活扩展

自定义节点的优势：

• 业务解耦：业务逻辑与流程引擎分离
• 灵活扩展：支持复杂业务场景
• 可复用：节点可在不同流程中复用
• 易维护：业务逻辑集中管理

SpringBoot：快速开发与集成

SpringBoot提供了：

• 自动配置：快速集成Flowable
• 注解支持：@EventListener等便捷注解
• AOP支持：便于统一处理

系统架构设计

我们的工作流引擎主要包括以下几个模块：

1. 流程定义管理：BPMN流程定义的CRUD操作
2. 流程实例管理：流程实例的启动和管理
3. 任务管理：待办任务的分配和处理
4. 自定义节点：业务逻辑的扩展点
5. 流程监控：流程执行状态监控
6. 权限控制：流程访问权限管理

核心实现思路

1. Flowable基础配置

@Configuration
@EnableProcessApplication
public class FlowableConfig {
    
    @Bean
    public ProcessEngineConfiguration processEngineConfiguration() {
        ProcessEngineConfiguration config = ProcessEngineConfiguration.createStandaloneProcessEngineConfiguration();
        
        // 数据库配置
        config.setJdbcUrl("jdbc:mysql://localhost:3306/flowable?useUnicode=true&characterEncoding=utf8");
        config.setJdbcUsername("root");
        config.setJdbcPassword("password");
        config.setJdbcDriver("com.mysql.cj.jdbc.Driver");
        
        // 自动创建表
        config.setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE);
        
        // 自定义配置
        config.setAsyncExecutorActivate(true); // 启用异步执行器
        config.setJobExecutorActivate(true);   // 启用作业执行器
        
        return config;
    }
    
    @Bean
    public ProcessEngine processEngine() {
        return processEngineConfiguration().buildProcessEngine();
    }
}

2. 流程定义服务

@Service
public class ProcessDefinitionService {
    
    @Autowired
    private RepositoryService repositoryService;
    
    @Autowired
    private RuntimeService runtimeService;
    
    /**
     * 部署流程定义
     */
    public String deployProcess(String processName, String bpmnXml) {
        Deployment deployment = repositoryService.createDeployment()
            .name(processName)
            .addString(processName + ".bpmn", bpmnXml)
            .deploy();
        
        return deployment.getId();
    }
    
    /**
     * 启动流程实例
     */
    public String startProcess(String processDefinitionKey, Map<String, Object> variables) {
        ProcessInstance processInstance = runtimeService.startProcessInstanceByKey(
            processDefinitionKey, variables);
        return processInstance.getId();
    }
    
    /**
     * 获取流程图
     */
    public InputStream getProcessImage(String processDefinitionId) {
        return repositoryService.getProcessDiagram(processDefinitionId);
    }
}

3. 任务管理服务

@Service
public class TaskService {
    
    @Autowired
    private org.flowable.task.api.TaskService taskService;
    
    @Autowired
    private RuntimeService runtimeService;
    
    /**
     * 获取待办任务
     */
    public List<TaskDto> getTodoTasks(String assignee) {
        List<org.flowable.task.api.Task> tasks = taskService.createTaskQuery()
            .taskAssignee(assignee)
            .orderByTaskCreateTime()
            .desc()
            .list();
        
        return tasks.stream()
            .map(this::convertToDto)
            .collect(Collectors.toList());
    }
    
    /**
     * 完成任务
     */
    public void completeTask(String taskId, Map<String, Object> variables) {
        taskService.complete(taskId, variables);
    }
    
    /**
     * 分配任务
     */
    public void assignTask(String taskId, String assignee) {
        taskService.setAssignee(taskId, assignee);
    }
    
    private TaskDto convertToDto(org.flowable.task.api.Task task) {
        TaskDto dto = new TaskDto();
        dto.setId(task.getId());
        dto.setName(task.getName());
        dto.setAssignee(task.getAssignee());
        dto.setCreateTime(task.getCreateTime());
        dto.setProcessInstanceId(task.getProcessInstanceId());
        dto.setProcessDefinitionId(task.getProcessDefinitionId());
        return dto;
    }
}

4. 请假流程示例

@Service
public class LeaveProcessService {
    
    @Autowired
    private ProcessDefinitionService processService;
    
    @Autowired
    private TaskService taskService;
    
    /**
     * 提交请假申请
     */
    public String submitLeaveRequest(LeaveRequest request) {
        Map<String, Object> variables = new HashMap<>();
        variables.put("employeeId", request.getEmployeeId());
        variables.put("employeeName", request.getEmployeeName());
        variables.put("leaveType", request.getLeaveType());
        variables.put("leaveDays", request.getLeaveDays());
        variables.put("startTime", request.getStartTime());
        variables.put("endTime", request.getEndTime());
        variables.put("reason", request.getReason());
        
        // 根据请假天数确定审批流程
        if (request.getLeaveDays() > 7) {
            variables.put("approvalLevel", "high");
        } else if (request.getLeaveDays() > 3) {
            variables.put("approvalLevel", "medium");
        } else {
            variables.put("approvalLevel", "low");
        }
        
        return processService.startProcess("leaveProcess", variables);
    }
    
    /**
     * 审批请假申请
     */
    public void approveLeave(String taskId, Boolean approved, String comment) {
        Map<String, Object> variables = new HashMap<>();
        variables.put("approved", approved);
        variables.put("comment", comment);
        
        taskService.completeTask(taskId, variables);
    }
}

自定义节点实现

1. 自定义业务逻辑节点

@Component
public class CustomBusinessNodeHandler implements JavaDelegate {
    
    @Autowired
    private NotificationService notificationService;
    
    @Autowired
    private EmailService emailService;
    
    @Override
    public void execute(DelegateExecution execution) {
        try {
            // 获取流程变量
            String employeeId = (String) execution.getVariable("employeeId");
            String processInstanceId = execution.getProcessInstanceId();
            
            // 执行业务逻辑
            executeBusinessLogic(employeeId, processInstanceId);
            
            // 发送通知
            notificationService.sendNotification("流程节点执行成功", employeeId);
            
            // 记录日志
            logExecution(execution);
            
        } catch (Exception e) {
            log.error("自定义节点执行失败", e);
            throw new BpmnError("BUSINESS_ERROR", "业务处理失败: " + e.getMessage());
        }
    }
    
    private void executeBusinessLogic(String employeeId, String processInstanceId) {
        // 具体的业务逻辑实现
        // 例如：更新数据库、调用外部服务等
    }
    
    private void logExecution(DelegateExecution execution) {
        // 记录执行日志
        String taskId = execution.getCurrentActivityId();
        String taskName = execution.getCurrentActivityName();
        
        log.info("自定义节点执行完成: taskId={}, taskName={}, processInstanceId={}", 
            taskId, taskName, execution.getProcessInstanceId());
    }
}

2. 动态任务分配节点

@Component
public class DynamicAssignmentNode implements JavaDelegate {
    
    @Autowired
    private UserService userService;
    
    @Override
    public void execute(DelegateExecution execution) {
        String leaveType = (String) execution.getVariable("leaveType");
        String employeeId = (String) execution.getVariable("employeeId");
        
        String approverId = determineApprover(leaveType, employeeId);
        
        // 设置下一个任务的处理人
        execution.setVariable("nextApprover", approverId);
        
        // 如果需要，可以在这里直接分配任务
        // taskService.assignTask(taskId, approverId);
    }
    
    private String determineApprover(String leaveType, String employeeId) {
        // 根据请假类型和员工信息确定审批人
        Employee employee = userService.getEmployeeById(employeeId);
        
        switch (leaveType) {
            case "annual":
                return employee.getManagerId(); // 年假找直属领导
            case "sick":
                return employee.getHrId(); // 病假找HR
            case "personal":
                return employee.getDirectorId(); // 事假找部门领导
            default:
                return employee.getManagerId();
        }
    }
}

3. 条件分支节点

@Component
public class LeaveConditionNode implements ExpressionCondition {
    
    @Override
    public boolean evaluate(DelegateExecution execution) {
        Integer leaveDays = (Integer) execution.getVariable("leaveDays");
        String approvalLevel = (String) execution.getVariable("approvalLevel");
        
        // 根据请假天数和审批级别决定流程走向
        if ("high".equals(approvalLevel)) {
            return true; // 走高级审批流程
        } else if (leaveDays > 3) {
            return true; // 超过3天走中等审批
        } else {
            return false; // 3天以内走简单审批
        }
    }
}

高级特性实现

1. 流程监听器

@Component
public class ProcessListener implements ExecutionListener {
    
    @Override
    public void notify(DelegateExecution execution) {
        String eventName = execution.getEventName();
        String processInstanceId = execution.getProcessInstanceId();
        String activityId = execution.getCurrentActivityId();
        
        switch (eventName) {
            case "start":
                log.info("流程开始: processInstanceId={}, activityId={}", 
                    processInstanceId, activityId);
                handleProcessStart(execution);
                break;
            case "end":
                log.info("流程结束: processInstanceId={}, activityId={}", 
                    processInstanceId, activityId);
                handleProcessEnd(execution);
                break;
            case "take":
                log.info("流程流转: processInstanceId={}, activityId={}", 
                    processInstanceId, activityId);
                handleFlowTransition(execution);
                break;
        }
    }
    
    private void handleProcessStart(DelegateExecution execution) {
        // 流程开始时的处理逻辑
        String businessKey = execution.getVariable("businessKey", String.class);
        // 记录流程开始日志、发送通知等
    }
    
    private void handleProcessEnd(DelegateExecution execution) {
        // 流程结束时的处理逻辑
        String processInstanceId = execution.getProcessInstanceId();
        // 更新业务状态、发送最终通知等
    }
    
    private void handleFlowTransition(DelegateExecution execution) {
        // 流程流转时的处理逻辑
        String transitionId = execution.getCurrentTransitionId();
        // 记录流转日志
    }
}

2. 任务监听器

@Component
public class TaskListener implements org.flowable.task.api.TaskListener {
    
    @Override
    public void notify(DelegateTask delegateTask) {
        String eventName = delegateTask.getEventName();
        
        switch (eventName) {
            case "create":
                handleTaskCreate(delegateTask);
                break;
            case "assignment":
                handleTaskAssignment(delegateTask);
                break;
            case "complete":
                handleTaskComplete(delegateTask);
                break;
        }
    }
    
    private void handleTaskCreate(DelegateTask delegateTask) {
        String assignee = delegateTask.getAssignee();
        String processInstanceId = delegateTask.getProcessInstanceId();
        
        // 任务创建时发送通知
        notificationService.sendTaskNotification(assignee, processInstanceId, delegateTask.getName());
    }
    
    private void handleTaskAssignment(DelegateTask delegateTask) {
        String oldAssignee = delegateTask.getVariable("oldAssignee", String.class);
        String newAssignee = delegateTask.getAssignee();
        
        // 任务重新分配时的处理
        if (!Objects.equals(oldAssignee, newAssignee)) {
            notificationService.sendReassignmentNotification(newAssignee, delegateTask.getName());
        }
    }
    
    private void handleTaskComplete(DelegateTask delegateTask) {
        String assignee = delegateTask.getAssignee();
        String processInstanceId = delegateTask.getProcessInstanceId();
        
        // 任务完成时的处理
        log.info("任务完成: assignee={}, processInstanceId={}", assignee, processInstanceId);
    }
}

3. 流程变量验证

@Component
public class ProcessVariableValidator {
    
    public ValidationResult validateVariables(String processKey, Map<String, Object> variables) {
        ValidationResult result = new ValidationResult();
        
        switch (processKey) {
            case "leaveProcess":
                validateLeaveVariables(variables, result);
                break;
            case "expenseProcess":
                validateExpenseVariables(variables, result);
                break;
            default:
                result.setValid(true);
        }
        
        return result;
    }
    
    private void validateLeaveVariables(Map<String, Object> variables, ValidationResult result) {
        String employeeId = (String) variables.get("employeeId");
        String leaveType = (String) variables.get("leaveType");
        Integer leaveDays = (Integer) variables.get("leaveDays");
        
        if (employeeId == null || employeeId.isEmpty()) {
            result.addError("employeeId", "员工ID不能为空");
        }
        
        if (leaveType == null || leaveType.isEmpty()) {
            result.addError("leaveType", "请假类型不能为空");
        }
        
        if (leaveDays == null || leaveDays <= 0) {
            result.addError("leaveDays", "请假天数必须大于0");
        } else if (leaveDays > 30) {
            result.addError("leaveDays", "请假天数不能超过30天");
        }
        
        result.setValid(result.getErrors().isEmpty());
    }
}

报销流程示例

报销流程定义（BPMN）

<?xml version="1.0" encoding="UTF-8"?>
<definitions xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xmlns:xsd="http://www.w3.org/2001/XMLSchema"
             xmlns:flowable="http://flowable.org/bpmn"
             typeLanguage="http://www.w3.org/2001/XMLSchema"
             expressionLanguage="http://www.w3.org/1999/XPath"
             targetNamespace="http://www.flowable.org/processdef">
             
    <process id="expenseProcess" name="报销流程" isExecutable="true">
        
        <startEvent id="startEvent" name="开始"/>
        
        <userTask id="fillExpense" name="填写报销单" flowable:assignee="${employeeId}">
            <extensionElements>
                <flowable:taskListener event="create" class="com.example.listener.TaskListener"/>
            </extensionElements>
        </userTask>
        
        <exclusiveGateway id="checkAmount" name="金额检查"/>
        
        <userTask id="deptApproval" name="部门领导审批" flowable:assignee="${deptManagerId}"/>
        <userTask id="financeApproval" name="财务审批" flowable:assignee="${financeManagerId}"/>
        <userTask id="ceoApproval" name="总经理审批" flowable:assignee="${ceoId}"/>
        
        <serviceTask id="paymentTask" name="支付处理" flowable:class="com.example.service.PaymentService"/>
        
        <endEvent id="endEvent" name="结束"/>
        
        <sequenceFlow id="flow1" sourceRef="startEvent" targetRef="fillExpense"/>
        <sequenceFlow id="flow2" sourceRef="fillExpense" targetRef="checkAmount"/>
        
        <sequenceFlow id="flow3" sourceRef="checkAmount" targetRef="deptApproval">
            <conditionExpression xsi:type="tFormalExpression">${expenseAmount <= 1000}</conditionExpression>
        </sequenceFlow>
        
        <sequenceFlow id="flow4" sourceRef="checkAmount" targetRef="financeApproval">
            <conditionExpression xsi:type="tFormalExpression">${expenseAmount > 1000 and expenseAmount <= 5000}</conditionExpression>
        </sequenceFlow>
        
        <sequenceFlow id="flow5" sourceRef="checkAmount" targetRef="ceoApproval">
            <conditionExpression xsi:type="tFormalExpression">${expenseAmount > 5000}</conditionExpression>
        </sequenceFlow>
        
        <sequenceFlow id="flow6" sourceRef="deptApproval" targetRef="paymentTask"/>
        <sequenceFlow id="flow7" sourceRef="financeApproval" targetRef="paymentTask"/>
        <sequenceFlow id="flow8" sourceRef="ceoApproval" targetRef="paymentTask"/>
        <sequenceFlow id="flow9" sourceRef="paymentTask" targetRef="endEvent"/>
        
    </process>
</definitions>

报销流程服务

@Service
public class ExpenseProcessService {
    
    @Autowired
    private ProcessDefinitionService processService;
    
    @Autowired
    private TaskService taskService;
    
    /**
     * 提交报销申请
     */
    public String submitExpenseRequest(ExpenseRequest request) {
        Map<String, Object> variables = new HashMap<>();
        variables.put("employeeId", request.getEmployeeId());
        variables.put("employeeName", request.getEmployeeName());
        variables.put("expenseAmount", request.getAmount());
        variables.put("expenseType", request.getExpenseType());
        variables.put("expenseDescription", request.getDescription());
        variables.put("expenseDate", request.getExpenseDate());
        variables.put("attachments", request.getAttachments());
        
        // 设置审批人
        Employee employee = userService.getEmployeeById(request.getEmployeeId());
        variables.put("deptManagerId", employee.getManagerId());
        variables.put("financeManagerId", employee.getFinanceManagerId());
        variables.put("ceoId", employee.getCeoId());
        
        return processService.startProcess("expenseProcess", variables);
    }
    
    /**
     * 审批报销申请
     */
    public void approveExpense(String taskId, Boolean approved, String comment) {
        Map<String, Object> variables = new HashMap<>();
        variables.put("approved", approved);
        variables.put("comment", comment);
        
        taskService.completeTask(taskId, variables);
    }
}

性能优化建议

1. 流程实例缓存

@Service
public class CachedProcessService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    public ProcessInstance getProcessInstance(String processInstanceId) {
        String cacheKey = "process:instance:" + processInstanceId;
        
        // 先从缓存获取
        ProcessInstance cached = (ProcessInstance) 
            redisTemplate.opsForValue().get(cacheKey);
        
        if (cached != null) {
            return cached;
        }
        
        // 从Flowable获取
        ProcessInstance instance = runtimeService.createProcessInstanceQuery()
            .processInstanceId(processInstanceId)
            .singleResult();
        
        // 缓存到Redis（缓存5分钟）
        if (instance != null) {
            redisTemplate.opsForValue().set(cacheKey, instance, Duration.ofMinutes(5));
        }
        
        return instance;
    }
}

2. 任务批量处理

@Service
public class BatchTaskService {
    
    public BatchTaskResult batchCompleteTasks(List<String> taskIds, Map<String, Object> variables) {
        List<CompletableFuture<Void>> futures = taskIds.stream()
            .map(taskId -> CompletableFuture.runAsync(() -> {
                taskService.completeTask(taskId, variables);
            }))
            .collect(Collectors.toList());
        
        try {
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            return BatchTaskResult.success(taskIds.size());
        } catch (Exception e) {
            return BatchTaskResult.failure("批量处理失败: " + e.getMessage());
        }
    }
}

安全考虑

1. 任务权限控制

@Service
public class TaskPermissionService {
    
    public boolean canAccessTask(String userId, String taskId) {
        org.flowable.task.api.Task task = taskService.createTaskQuery()
            .taskId(taskId)
            .singleResult();
        
        if (task == null) {
            return false;
        }
        
        // 检查用户是否有权限访问该任务
        String taskAssignee = task.getAssignee();
        String taskCandidateUser = task.getOwner();
        
        return userId.equals(taskAssignee) || userId.equals(taskCandidateUser);
    }
    
    public boolean canCompleteTask(String userId, String taskId) {
        // 检查用户权限
        if (!canAccessTask(userId, taskId)) {
            return false;
        }
        
        // 检查任务状态
        org.flowable.task.api.Task task = taskService.createTaskQuery()
            .taskId(taskId)
            .singleResult();
        
        return task != null && task.getAssignee() != null;
    }
}

2. 流程数据安全

@Component
public class ProcessDataSecurityService {
    
    public Map<String, Object> sanitizeProcessVariables(Map<String, Object> variables) {
        // 移除敏感信息
        Map<String, Object> sanitized = new HashMap<>(variables);
        
        // 移除系统变量
        sanitized.remove("password");
        sanitized.remove("secret");
        sanitized.remove("token");
        
        // 对敏感数据进行脱敏
        if (sanitized.containsKey("bankAccount")) {
            String account = (String) sanitized.get("bankAccount");
            sanitized.put("bankAccount", maskBankAccount(account));
        }
        
        return sanitized;
    }
    
    private String maskBankAccount(String account) {
        if (account == null || account.length() < 8) {
            return account;
        }
        return account.substring(0, 4) + "****" + account.substring(account.length() - 4);
    }
}

监控与运维

1. 流程监控

@Component
public class ProcessMonitorService {
    
    private final MeterRegistry meterRegistry;
    
    public void recordProcessStart(String processKey) {
        Counter.builder("process_started_total")
            .tag("process_key", processKey)
            .register(meterRegistry)
            .increment();
    }
    
    public void recordProcessComplete(String processKey, long duration) {
        Timer.builder("process_duration_seconds")
            .tag("process_key", processKey)
            .register(meterRegistry)
            .record(duration, TimeUnit.MILLISECONDS);
    }
    
    public void recordTaskCompletion(String processKey, String taskName) {
        Counter.builder("task_completed_total")
            .tag("process_key", processKey)
            .tag("task_name", taskName)
            .register(meterRegistry)
            .increment();
    }
}

2. 流程状态查询

@Service
public class ProcessQueryService {
    
    @Autowired
    private HistoryService historyService;
    
    public ProcessInstanceHistory getProcessHistory(String processInstanceId) {
        HistoricProcessInstance historicInstance = historyService
            .createHistoricProcessInstanceQuery()
            .processInstanceId(processInstanceId)
            .singleResult();
        
        List<HistoricActivityInstance> activityInstances = historyService
            .createHistoricActivityInstanceQuery()
            .processInstanceId(processInstanceId)
            .orderByHistoricActivityInstanceStartTime()
            .asc()
            .list();
        
        ProcessInstanceHistory history = new ProcessInstanceHistory();
        history.setProcessInstanceId(processInstanceId);
        history.setStartTime(historicInstance.getStartTime());
        history.setEndTime(historicInstance.getEndTime());
        history.setDuration(historicInstance.getDurationInMillis());
        history.setActivities(activityInstances.stream()
            .map(this::convertToActivityDto)
            .collect(Collectors.toList()));
        
        return history;
    }
    
    private ActivityDto convertToActivityDto(HistoricActivityInstance activity) {
        ActivityDto dto = new ActivityDto();
        dto.setActivityId(activity.getActivityId());
        dto.setActivityName(activity.getActivityName());
        dto.setActivityType(activity.getActivityType());
        dto.setStartTime(activity.getStartTime());
        dto.setEndTime(activity.getEndTime());
        dto.setDuration(activity.getDurationInMillis());
        return dto;
    }
}

最佳实践

1. 流程设计规范

• 单一职责：每个流程专注于一个业务场景
• 状态明确：流程状态清晰易懂
• 异常处理：完善的异常处理机制
• 可回退：支持流程回退操作

2. 自定义节点设计

• 职责单一：每个自定义节点只处理一种业务逻辑
• 可配置：支持通过配置参数控制行为
• 可测试：便于单元测试和集成测试
• 可监控：提供执行状态和指标

3. 性能优化

• 异步处理：长时间操作异步执行
• 批量操作：支持批量任务处理
• 连接池：合理配置数据库连接池
• 缓存策略：合理使用缓存减少数据库访问

总结

通过SpringBoot + Flowable + 自定义节点的组合，我们可以构建一个企业级的工作流引擎。关键在于：

1. 标准化：使用BPMN 2.0标准定义流程
2. 灵活扩展：通过自定义节点支持复杂业务逻辑
3. 可视化：支持流程图查看和设计
4. 安全可靠：完善的权限控制和异常处理
5. 性能优异：支持高并发流程处理

记住，工作流引擎不是一蹴而就的，需要根据业务特点持续优化。掌握了这些技巧，你就能构建一个灵活高效的工作流系统，让业务流程管理更简单！