Flowable并行网关案例

wen java案例 2026-08-02 2

本文目录导读：

Flowable并行网关案例

业务流程设计
BPMN模型XML定义
Java实现代码
配置与最佳实践

我为您设计一个完整的Flowable并行网关实战案例,涵盖请假审批和采购会签两个典型场景。

业务流程设计

场景1：员工请假审批（多级并行+条件分支）

开始 → 提交请假申请 → 【并行网关】 
  ├─→ 部门经理审批（并行）
  ├─→ HR备案（并行）  
  └─→ 项目经理审批（并行）
  → 【并行网关汇聚】 → 判断天数>3? 
  ├─ 是 → 总经理审批 → 结束
  └─ 否 → 直接结束

场景2：采购合同会签（多节点并行+投票）

开始 → 提交合同 → 【并行网关】 
  ├─→ 法务审批（并行）
  ├─→ 财务审批（并行）
  ├─→ 技术评审（并行）
  └─→ 采购主管审批（并行）
  → 【并行汇聚+计数器】 → 至少3个通过 → 总经理审批 → 结束

BPMN模型XML定义

<?xml version="1.0" encoding="UTF-8"?>
<definitions xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL"
             xmlns:flowable="http://flowable.org/bpmn"
             targetNamespace="http://flowable.org/bpmn">
  <!-- 请假流程 -->
  <process id="leaveProcess" name="请假审批流程" isExecutable="true">
    <startEvent id="startEvent" name="开始"/>
    <userTask id="submitLeave" name="提交请假申请" flowable:assignee="${applyUser}"/>
    <parallelGateway id="parallelStart" name="并行开始"/>
    <userTask id="deptManagerApprove" name="部门经理审批" flowable:assignee="${deptManager}"/>
    <userTask id="hrRecord" name="HR备案" flowable:assignee="${hrUser}"/>
    <userTask id="projectManagerApprove" name="项目经理审批" flowable:assignee="${projectManager}"/>
    <parallelGateway id="parallelEnd" name="并行汇聚"/>
    <exclusiveGateway id="daysCheck" name="天数判断"/>
    <userTask id="gmApprove" name="总经理审批" flowable:assignee="generalManager"/>
    <endEvent id="endEvent" name="结束"/>
    <!-- 流程连线 -->
    <sequenceFlow id="flow1" sourceRef="startEvent" targetRef="submitLeave"/>
    <sequenceFlow id="flow2" sourceRef="submitLeave" targetRef="parallelStart"/>
    <sequenceFlow id="flow3" sourceRef="parallelStart" targetRef="deptManagerApprove"/>
    <sequenceFlow id="flow4" sourceRef="parallelStart" targetRef="hrRecord"/>
    <sequenceFlow id="flow5" sourceRef="parallelStart" targetRef="projectManagerApprove"/>
    <sequenceFlow id="flow6" sourceRef="deptManagerApprove" targetRef="parallelEnd"/>
    <sequenceFlow id="flow7" sourceRef="hrRecord" targetRef="parallelEnd"/>
    <sequenceFlow id="flow8" sourceRef="projectManagerApprove" targetRef="parallelEnd"/>
    <sequenceFlow id="flow9" sourceRef="parallelEnd" targetRef="daysCheck"/>
    <sequenceFlow id="flow10" sourceRef="daysCheck" targetRef="gmApprove">
      <conditionExpression xsi:type="tFormalExpression">
        <![CDATA[${days > 3}]]>
      </conditionExpression>
    </sequenceFlow>
    <sequenceFlow id="flow11" sourceRef="daysCheck" targetRef="endEvent">
      <conditionExpression xsi:type="tFormalExpression">
        <![CDATA[${days <= 3}]]>
      </conditionExpression>
    </sequenceFlow>
    <sequenceFlow id="flow12" sourceRef="gmApprove" targetRef="endEvent"/>
  </process>
</definitions>

Java实现代码

并行网关部署与启动

package com.example.flowable;
import org.flowable.engine.*;
import org.flowable.engine.repository.Deployment;
import org.flowable.engine.runtime.ProcessInstance;
import org.flowable.task.api.Task;
import org.flowable.variable.VariableMap;
import org.flowable.variable.Variables;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@Service
public class ParallelGatewayService {
    private final ProcessEngine processEngine;
    private final RuntimeService runtimeService;
    private final TaskService taskService;
    private final HistoryService historyService;
    public ParallelGatewayService(ProcessEngine processEngine) {
        this.processEngine = processEngine;
        this.runtimeService = processEngine.getRuntimeService();
        this.taskService = processEngine.getTaskService();
        this.historyService = processEngine.getHistoryService();
    }
    /**
     * 部署流程定义
     */
    @PostConstruct
    public void deployProcess() {
        // 部署BPMN文件 - 从类路径加载
        Deployment deployment = processEngine.getRepositoryService()
            .createDeployment()
            .addClasspathResource("processes/leave-process.bpmn20.xml")
            .name("请假审批流程")
            .deploy();
        System.out.println("部署ID: " + deployment.getId());
    }
    /**
     * 启动请假流程
     */
    public ProcessInstance startLeaveProcess(String applyUser, 
                                            String deptManager,
                                            String hrUser, 
                                            String projectManager,
                                            int days) {
        Map<String, Object> variables = new HashMap<>();
        variables.put("applyUser", applyUser);
        variables.put("deptManager", deptManager);
        variables.put("hrUser", hrUser);
        variables.put("projectManager", projectManager);
        variables.put("days", days);
        ProcessInstance processInstance = runtimeService
            .startProcessInstanceByKey("leaveProcess", variables);
        System.out.println("流程实例ID: " + processInstance.getId());
        return processInstance;
    }
    /**
     * 查询待办任务
     */
    public List<Task> getTasksByAssignee(String assignee) {
        return taskService.createTaskQuery()
            .taskAssignee(assignee)
            .orderByTaskCreateTime()
            .desc()
            .list();
    }
    /**
     * 完成任务并设置流程变量
     */
    public void completeTask(String taskId, Map<String, Object> variables) {
        taskService.complete(taskId, variables);
        System.out.println("任务完成: " + taskId);
    }
    /**
     * 获取活动任务
     */
    public void getActiveTasks(String processInstanceId) {
        List<Task> tasks = taskService.createTaskQuery()
            .processInstanceId(processInstanceId)
            .active()
            .list();
        System.out.println("当前活动任务数量: " + tasks.size());
        tasks.forEach(task -> {
            System.out.println("任务ID: " + task.getId() + 
                             ", 名称: " + task.getName() + 
                             ", 处理人: " + task.getAssignee());
        });
    }
}

并行网关测试控制器

package com.example.flowable.controller;
import com.example.flowable.ParallelGatewayService;
import org.flowable.engine.runtime.ProcessInstance;
import org.flowable.task.api.Task;
import org.springframework.web.bind.annotation.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@RestController
@RequestMapping("/api/parallel")
public class ParallelGatewayController {
    private final ParallelGatewayService parallelGatewayService;
    public ParallelGatewayController(ParallelGatewayService service) {
        this.parallelGatewayService = service;
    }
    /**
     * 启动请假流程
     */
    @PostMapping("/start")
    public Map<String, Object> startLeaveProcess(@RequestBody Map<String, Object> params) {
        ProcessInstance processInstance = parallelGatewayService.startLeaveProcess(
            (String) params.get("applyUser"),
            (String) params.get("deptManager"),
            (String) params.get("hrUser"),
            (String) params.get("projectManager"),
            (Integer) params.get("days")
        );
        Map<String, Object> result = new HashMap<>();
        result.put("processInstanceId", processInstance.getId());
        result.put("message", "流程启动成功");
        return result;
    }
    /**
     * 查询某人的待办任务
     */
    @GetMapping("/tasks/{assignee}")
    public List<Task> getTasks(@PathVariable String assignee) {
        return parallelGatewayService.getTasksByAssignee(assignee);
    }
    /**
     * 完成任务
     */
    @PostMapping("/complete/{taskId}")
    public Map<String, Object> completeTask(@PathVariable String taskId,
                                           @RequestBody(required = false) Map<String, Object> variables) {
        parallelGatewayService.completeTask(taskId, variables);
        Map<String, Object> result = new HashMap<>();
        result.put("success", true);
        result.put("message", "任务完成");
        return result;
    }
    /**
     * 查看流程当前状态
     */
    @GetMapping("/status/{processInstanceId}")
    public Map<String, Object> getProcessStatus(@PathVariable String processInstanceId) {
        parallelGatewayService.getActiveTasks(processInstanceId);
        Map<String, Object> result = new HashMap<>();
        result.put("processInstanceId", processInstanceId);
        result.put("status", "running");
        return result;
    }
}

并行网关高级用法

package com.example.flowable.advanced;
import org.flowable.engine.RuntimeService;
import org.flowable.engine.TaskService;
import org.flowable.engine.runtime.ProcessInstance;
import org.flowable.task.api.Task;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.*;
@Service
public class AdvancedParallelGateway {
    private final RuntimeService runtimeService;
    private final TaskService taskService;
    public AdvancedParallelGateway(RuntimeService runtimeService, TaskService taskService) {
        this.runtimeService = runtimeService;
        this.taskService = taskService;
    }
    /**
     * 动态并行子流程
     */
    @Transactional
    public void startDynamicParallelProcess(int numberOfBranches) {
        ProcessInstance processInstance = runtimeService
            .startProcessInstanceByKey("dynamicParallelProcess");
        // 动态创建并行分支
        for (int i = 0; i < numberOfBranches; i++) {
            runtimeService.createChangeActivityStateBuilder()
                .processInstanceId(processInstance.getId())
                .moveActivityIdTo("parallelStart", "dynamicTask" + i)
                .changeState();
        }
    }
    /**
     * 并行任务的异步处理
     */
    public void handleParallelTasksAsync(String processInstanceId) {
        List<Task> parallelTasks = taskService.createTaskQuery()
            .processInstanceId(processInstanceId)
            .list();
        // 使用并行流处理任务
        parallelTasks.parallelStream().forEach(task -> {
            Map<String, Object> variables = new HashMap<>();
            variables.put("task_" + task.getId() + "_approved", true);
            variables.put("task_" + task.getId() + "_time", new Date());
            taskService.complete(task.getId(), variables);
        });
    }
    /**
     * 带条件的并行网关
     */
    public void conditionalParallelGateway(String processInstanceId) {
        // 检查各个并行分支的状态
        Map<String, Object> variables = runtimeService.getVariables(processInstanceId);
        // 统计通过数量
        long approvedCount = variables.entrySet().stream()
            .filter(entry -> entry.getKey().endsWith("_approved"))
            .filter(entry -> Boolean.TRUE.equals(entry.getValue()))
            .count();
        System.out.println("已通过的任务数量: " + approvedCount);
    }
    /**
     * 并行网关中的定时器
     */
    public void parallelWithTimer(String processInstanceId) {
        // 设置并行任务的超时时间
        Map<String, Object> timerVariables = new HashMap<>();
        timerVariables.put("timerDuration", "PT15M"); // 15分钟超时
        // 设置定时器任务
        runtimeService.createChangeActivityStateBuilder()
            .processInstanceId(processInstanceId)
            .moveActivityIdTo("parallelGateway", "timerBoundary")
            .processInstanceVariable("timerDuration", timerVariables.get("timerDuration"))
            .changeState();
    }
}

完整测试类

package com.example.flowable;
import org.flowable.engine.*;
import org.flowable.engine.repository.Deployment;
import org.flowable.engine.runtime.ProcessInstance;
import org.flowable.task.api.Task;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@SpringBootTest
public class ParallelGatewayTest {
    @Autowired
    private ProcessEngine processEngine;
    @Test
    public void testParallelGateway() {
        // 1. 部署流程
        Deployment deployment = processEngine.getRepositoryService()
            .createDeployment()
            .addClasspathResource("processes/leave-process.bpmn20.xml")
            .deploy();
        System.out.println("部署ID: " + deployment.getId());
        // 2. 启动流程
        Map<String, Object> variables = new HashMap<>();
        variables.put("applyUser", "张三");
        variables.put("deptManager", "李经理");
        variables.put("hrUser", "HR小王");
        variables.put("projectManager", "项目经理");
        variables.put("days", 5);
        ProcessInstance processInstance = processEngine.getRuntimeService()
            .startProcessInstanceByKey("leaveProcess", variables);
        System.out.println("流程实例ID: " + processInstance.getId());
        // 3. 查询待办任务
        List<Task> tasks = processEngine.getTaskService()
            .createTaskQuery()
            .processInstanceId(processInstance.getId())
            .list();
        System.out.println("启动后任务数: " + tasks.size());
        // 4. 模拟并行审批
        tasks.forEach(task -> {
            System.out.println("处理任务: " + task.getName() + " (ID: " + task.getId() + ")");
            Map<String, Object> taskVariables = new HashMap<>();
            taskVariables.put("approved", true);
            taskVariables.put("approver", task.getAssignee());
            processEngine.getTaskService().complete(task.getId(), taskVariables);
        });
        // 5. 检查并行网关后的分支
        List<Task> afterParallelTasks = processEngine.getTaskService()
            .createTaskQuery()
            .processInstanceId(processInstance.getId())
            .list();
        System.out.println("并行汇聚后任务数: " + afterParallelTasks.size());
        // 6. 根据条件变量判断流程走向
        if (variables.get("days").toString().equals("5")) {
            System.out.println("请假超过3天，需要总经理审批");
        }
        // 7. 完成剩余任务
        for (Task task : afterParallelTasks) {
            processEngine.getTaskService().complete(task.getId());
        }
        // 8. 验证流程是否结束
        ProcessInstance endedProcess = processEngine.getRuntimeService()
            .createProcessInstanceQuery()
            .processInstanceId(processInstance.getId())
            .singleResult();
        if (endedProcess == null) {
            System.out.println("流程已正常结束");
        }
    }
}

配置与最佳实践

application.yml配置

flowable:
  # 自动部署BPMN文件
  async-executor-activate: true
  database-schema-update: true
  # 历史级别配置
  history-level: audit
  # 流程定义缓存
  process-definition-cache-limit: 100
  # 自定义表达式
  custom-expression:
    myBean: com.example.flowable.CustomBean

并行网关使用注意事项

避免死锁：确保所有并行分支最终都会到达汇聚网关
变量传递：合理使用流程变量控制并行分支
性能优化：大并发场景使用异步执行器
错误处理：为每个并行分支配置错误边界事件
监控：使用Flowable Cockpit监控并行执行状态

这个完整的案例展示了Flowable并行网关的核心用法,可以根据实际业务需求进行扩展和调整。