云环境中 Flowable 与其他服务的交互实践

单体应用中，流程引擎与业务代码在同一个进程里，事件监听、状态更新都可以直接调用。到了云环境（微服务架构）下，Flowable 通常是独立部署的流程引擎服务，与业务服务、用户/组织服务、待办服务等各自独立，服务之间只能通过远程通道通信。交互方式就变成了两件事：

1. 命令通道（HTTP）：业务服务调用流程引擎服务，完成"启动流程、查询当前任务"等操作；
2. 事件通道（消息队列）：流程引擎把"流程启动、任务创建、审批通过、驳回"等事件通过 RocketMQ 广播给业务服务与待办服务，业务方按需消费。

按职责，这套交互分为两个部分：

部分	角色
流程引擎服务	内置 Flowable 引擎，提供 HTTP 接口并向外发送消息
客户端 SDK	供业务服务引入，封装远程调用与消息消费的粘合代码

总体架构

两条通道的分工很清晰：

• HTTP：业务发起动作（启动流程、查询任务状态），需要同步拿到结果；
• RocketMQ：引擎状态变化（流程启动、任务创建、审批结果），异步通知，业务方消费后驱动自己的业务（更新单据状态、生成待办、发通知）。

启动流程：HTTP 命令通道

客户端：HTTP Interface 声明式调用

业务服务引入客户端 SDK 后，通过 Spring HTTP Interface（@HttpExchange）声明远程接口。ProcessInstanceStartClient 指向流程引擎服务：

@HttpProxy(Constant.SERVICE_NAME)
@HttpExchange("/api/client/process-instance")
public interface ProcessInstanceStartClient {

    /**
     * 启动流程实例，请务必在事务提交之后才调用此方法，
     * 否则可能会出现数据不一致的情况。
     */
    @PostExchange("/start")
    ResponseBean<ProcessInstanceStartVo> start(@RequestBody ProcessInstanceStartDto startDto);
}

@HttpProxy(Constant.SERVICE_NAME) 负责服务发现与负载均衡（SERVICE_NAME 指向流程引擎服务），业务服务里只要注入这个接口即可发起调用，不需要关心 HTTP 细节。

服务端：流程启动的完整处理

服务端由 ProcessInstanceStartClientController 接收请求，转交 ProcessInstanceStartServiceImpl。启动逻辑包含几个关键环节：

@Transactional(rollbackFor = Exception.class)
@Override
public ProcessInstanceStartVo start(ProcessInstanceStartDto startDto) {
    // 1. 根据业务类型解析流程配置
    BusinessProcessConfig config = this.businessProcessModelService.findByBusinessMark(startDto.getBusinessMark());
    if (config == null || StringUtils.isEmpty(config.getProcessDefinitionId())) {
        throw new ServiceException(ProcessInstanceError.NO_PROCESS_CONFIG);
    }

    // 2. 防重复启动：同一业务键、同一业务类型只允许一个进行中的流程
    long processCount = this.countProcessInstance(startDto.getBusinessKey(), startDto.getBusinessMark());
    if (processCount > 0) {
        throw new ServiceException(ProcessInstanceError.PROCESS_INSTANCE_STARTED);
    }

    // 3. 组装流程变量：组织架构 + 业务类型 + 租户 + 发起人 + 表单
    Map<String, Object> variableMap = ...;
    variableMap.put(CommonVariableNames.STORE_ID, startDto.getStoreId());
    variableMap.put(CommonVariableNames.REGION_ID, startDto.getRegionId());
    variableMap.put(CommonVariableNames.DEPT_ID, startDto.getDeptId());
    variableMap.put(CommonVariableNames.BUSINESS_MARK, startDto.getBusinessMark());
    variableMap.put(CommonVariableNames.MESSAGE_TEMPLATE_ID, config.getMessageTemplateId());
    variableMap.put(CommonVariableNames.TENANT_ID, SecurityUtils.getTenantId());
    variableMap.put(CommonVariableNames.INITIATOR, starter);
    variableMap.put(CommonVariableNames.STARTER, starter);

    // 4. 通过 builder 启动流程，覆盖流程定义的租户
    ProcessInstance processInstance = processEngine.getRuntimeService()
            .createProcessInstanceBuilder()
            .processDefinitionId(config.getProcessDefinitionId())
            .businessKey(startDto.getBusinessKey())
            .overrideProcessDefinitionTenantId(SecurityUtils.getTenantId())
            .variables(variableMap)
            .start();

    // 5. 保存表单数据
    this.saveFormData(startDto, processInstance);
    ...
}

值得注意的设计点：

• 业务类型（businessMark）路由：业务服务不直接指定流程定义 ID，而是传 businessMark，由流程引擎服务根据配置（BusinessProcessConfig：流程定义 ID + 消息模板 ID）解析。这样业务侧与流程模型解耦，换流程版本不用改业务代码；
• 防重复启动：用 businessKey + businessMark 查询进行中的流程实例，存在则拒绝，保证"一笔单据只能有一条审批流"；
• 多租户：overrideProcessDefinitionTenantId(tenantId) 让启动的流程实例归属当前租户，同时租户 ID 也作为流程变量写入，后续所有事件、消息都能带上租户上下文；
• 发起人：通过 Authentication.setAuthenticatedUserId(starter) 写入引擎上下文，使引擎记录的 startUserId 与操作者一致。

查询当前任务

同样通过 HTTP 通道：TaskQueryClient 批量查询流程实例的当前任务（taskCandidateOrAssigned 匹配候选人/办理人，active() 过滤已结束任务），业务列表页可以直接拿到"当前审批人是谁、谁可以审批"等信息：

List<Task> tasks = processEngine.getTaskService()
        .createTaskQuery()
        .processInstanceIdIn(processInstanceIds)
        .taskCandidateOrAssigned(userId)
        .active()
        .includeIdentityLinks()
        .list();

事件监听：引擎状态变化的起点

流程引擎服务内部，Flowable 事件是整个交互链路的起点。所有引擎事件监听器在应用启动时批量注册（见 FlowableListenerConfiguration，注册逻辑在 ApplicationStartedEvent 中完成）：

for (AbstractFlowableEngineEventListener<? extends FlowableEvent> listener : flowableEngineEventListeners) {
    this.runtimeService.addEventListener(listener);
}

监听器统一继承 AbstractFlowableEngineEventListener<T>，基类负责类型判断与强转，子类只需声明关注的事件类型并实现业务处理：

public class ProcessInstanceStartedListener extends AbstractProcessInstanceListener<FlowableProcessStartedEvent> {
    @Override
    protected void onEngineEvent(FlowableProcessStartedEvent engineEvent) {
        // 1. 查询流程实例（含变量）
        ProcessInstance processInstance = ...;
        // 2. 组装消息并通过 MessageSender 发送给业务服务
        ProcessInstanceEventMessage msg = this.createProcessInstanceEventMessage(processInstance);
        Topic topic = new Topic(Topics.PROCESS_INSTANCE_EVENT_TOPIC, msg.getBusinessMark());
        messageSender.send(topic, processInstance.getProcessInstanceId(), msg, this.getUserContext(processInstance));
        // 3. 发布待办创建 Spring 事件，由待办监听器转成待办消息
        ProcessInstanceCreateTodoEvent event = new ProcessInstanceCreateTodoEvent(this, processInstance);
        this.applicationEventPublisher.publishEvent(event);
    }

    @Override
    public Collection<? extends FlowableEventType> getTypes() {
        return Set.of(FlowableEngineEventType.PROCESS_STARTED);
    }
}

流程引擎服务实际监听的引擎事件与业务动作对应关系：

引擎事件	监听器	业务动作
PROCESS_STARTED	ProcessInstanceStartedListener	发送流程启动消息（状态=审批中）、创建流程待办
PROCESS_COMPLETED	ProcessInstanceCompletedListener	发送流程完成消息（状态=通过/驳回）、发布审批待办完成事件
PROCESS_CANCELLED	ProcessInstanceCancelledListener	发送流程撤回消息（状态=撤回/驳回）、发布撤回待办事件
TASK_CREATED	TaskCreatedEventListener	分配候选人、无审批人自动通过、发送任务创建消息、创建任务待办
TASK_COMPLETED	TaskCompleteEventListener	发送任务完成消息（携带表单数据）、发布任务审批通过待办事件
自定义 TASK_REJECT	TaskRejectedEventListener	发送任务驳回消息、发布任务驳回待办事件

监听器做的事归纳为三类：发消息给业务服务、发布应用内 Spring 事件驱动待办、直接操作引擎（如自动通过）。其中"发消息"与"待办"是与其他服务交互的两个出口。

发送消息：RocketMQ 事件通道

消息发送抽象

流程引擎服务通过统一的 MessageSender 发送消息，Topic 封装了"主题 + 业务标签"两层信息：

Topic topic = new Topic(Topics.PROCESS_INSTANCE_EVENT_TOPIC, msg.getBusinessMark());
messageSender.send(topic, processInstance.getProcessInstanceId(), msg, this.getUserContext(processInstance));

主题定义集中在 Topics 中：

public class Topics {
    public static final String PROCESS_INSTANCE_EVENT_TOPIC = "ProcessInstanceEventMessage";
    public static final String TASK_EVENT_TOPIC = "TaskEventMessage";
}

两类业务消息体

流程实例事件消息（ProcessInstanceEventMessage）：

@Data
public class ProcessInstanceEventMessage {
    private String processInstanceId;   // 流程实例ID
    private String businessKey;         // 业务ID
    private String businessMark;        // 业务类型
    private String starter;             // 发起人
    private ProcessInstanceState processInstanceState; // 流程状态
    private String tenantId;            // 租户ID
    private LocalDateTime endTime;      // 流程结束时间
}

任务事件消息（TaskEventMessage）：

@Data
public class TaskEventMessage {
    private String processInstanceId;
    private String businessKey;
    private String businessMark;
    private String taskId;
    private String taskName;
    private String taskDefinitionKey;
    private Map<String, Object> taskVariables;
    private String formKey;
    private Map<String, Object> formData;   // 审批表单数据
    private TaskState state;                // 任务状态（审批中/通过/驳回）
}

消息体的设计有两个要点：

• 业务无关：消息只携带流程/任务的关键标识与状态，不含具体业务字段，业务数据仍留在业务服务自己的库里，业务方通过 businessKey 反查；
• 状态归一：processInstanceState / state 统一为枚举（PROCESSING/APPROVE/CANCEL/REJECT），业务方不需要关心 Flowable 内部术语。

待办消息

待办联动是另一路消息。UserTaskTodoEventListener 监听应用内的待办 Spring 事件（TaskCreateTodoEvent、TaskApproveTodoEvent、TaskRejectTodoEvent），转换为待办服务的消息（TodoCreateEvent 等结构由待办服务 SDK 提供）：

@EventListener(TaskCreateTodoEvent.class)
private void onCreate(TaskCreateTodoEvent event) {
    ...
    TodoCreateEvent createEvent = this.createTodoCreateEvent(processInstance);
    createEvent.setTodoType(TodoTypeEnums.TODO);
    createEvent.setToUser(new ArrayList<>(event.getUserIds()));
    createEvent.setTaskId(taskInfo.getId());
    createEvent.setTenantId(processInstance.getTenantId());
    createEvent.setBusinessState(TaskState.PROCESSING.getValue());
    Topic topic = new Topic(TopicConstant.TODO_CREATE, createEvent.getTodoType().name());
    this.messageSender.send(topic, createEvent.getBusinessId(), createEvent);
}

这里体现了"Spring 事件（进程内）→ RocketMQ 消息（跨服务）"的两级转换：引擎事件监听器不直接依赖待办服务，而是发布 TodoEvent，由专门的待办监听器负责与待办服务对接，职责单一、便于替换。

业务侧消费：客户端 SDK 的粘合

业务服务引入客户端 SDK 后，SDK 通过 WorkflowAutoConfiguration 自动装配两个 RocketMQ 消费者，并收集业务方注册的所有业务监听器：

@AutoConfiguration
public class WorkflowAutoConfiguration {

    @Bean
    public ProcessInstanceEventListener processInstanceEventListener(
            @Autowired(required = false) List<BaseProcessInstanceListener> listeners) {
        return new ProcessInstanceEventListener(listeners);
    }

    @Bean
    public TaskEventListener taskEventListener(List<BaseTaskListener> listeners) {
        return new TaskEventListener(listeners);
    }
}

消费逻辑由 SDK 提供，业务方只需要实现监听接口。ProcessInstanceEventListener（RocketMQ 消费者）收到消息后按 businessMark 路由：

@RocketMQMessageListener(topic = Topics.PROCESS_INSTANCE_EVENT_TOPIC,
        consumerGroup = "ProcessInstanceEventListener_${spring.application.name}")
public class ProcessInstanceEventListener extends AbstractRocketMQListener<ProcessInstanceEventMessage> {

    @Override
    public void handle(MessageContext<ProcessInstanceEventMessage> message) {
        ProcessInstanceEventMessage eventMessage = message.getBody();
        for (BaseProcessInstanceListener listener : listeners) {
            if (listener.isSupported(eventMessage)) {
                listener.execute(eventMessage);
            }
        }
    }
}

isSupported 默认按 businessMark 匹配，SDK 还提供了按流程/任务状态二次过滤的语义接口：

public interface ProcessInstanceStartedListener extends BaseProcessInstanceListener {
    @Override
    default boolean isSupported(ProcessInstanceEventMessage message) {
        return BaseProcessInstanceListener.super.isSupported(message)
                && ProcessInstanceState.PROCESSING.equals(message.getProcessInstanceState());
    }

    default void execute(ProcessInstanceEventMessage message) {
        this.onStarted(message);
    }

    void onStarted(ProcessInstanceEventMessage message);
}

同理 TaskEventListener 消费任务消息后按 state 分发到 onCreated / onApproved / onRejected。业务服务的开发者只需要：

1. 实现 ProcessInstanceStartedListener / ProcessInstanceApproveListener 等接口，声明自己的 businessMark；
2. 在 onStarted / onApproved / onRejected 中更新单据状态、发通知；
3. 其余（MQ 消费、路由、反序列化）全部由 SDK 完成。

核心业务流程闭环

以一次完整的审批为例，串联所有通道：

跨服务反向调用

流程引擎服务在运行过程中也会调用其他服务。典型场景是抄送任务：DefaultCopyTaskDelegate（服务任务委托）在执行抄送时调用用户/组织服务查询组织范围内的人员：

DataScopeUserQueryDto queryDto = new DataScopeUserQueryDto();
queryDto.setDeptId(deptId);
queryDto.setStoreId(storeId);
queryDto.setPositions(groups.getPositions());
queryDto.setRoles(groups.getRoles());
queryDto.setGroupScope(groupScope);
queryDto.setTenantId(tenantId);
List<FlatUserVo> data = HttpInterfaceUtils.getData(this.userHttpInterface.queryScopeUser(queryDto));

抄送人员的解析支持三种来源：流程变量表达式（${contextUser}）、固定用户 ID、按组织岗位/角色查询；查到的用户集合通过待办事件发给待办服务。这展示了云环境下服务间互相调用的常态：流程引擎服务既是消息生产者，也是其他服务的消费者。

云环境设计要点

1. 多租户贯穿

租户上下文（tenantId）贯穿整条链路：

• 启动时 overrideProcessDefinitionTenantId + 写入 TENANT_ID 流程变量；
• 消息体携带 tenantId，业务侧可按租户隔离处理；
• 跨服务查询（用户/组织、第三方配置）都带上 tenantId。

2. 事务边界

跨服务交互最大的风险是"消息发出去了，事务还没提交"：

• ProcessInstanceStartClient 的注释明确要求事务提交之后再调用启动接口；
• TaskTransformServiceImpl（转办）在 afterCommit 回调中发布事件，保证事务成功后才通知；
• TaskCreatedEventListener 对自动通过后的任务处理做了延迟 5 秒的调度，避免发起方事务未提交而审批通过消息已到达业务服务。

3. 幂等与防重

• 启动流程前用 businessKey + businessMark 校验，防止重复发起；
• 消息消费侧（RocketMQ 重试、接口重试）要求业务监听器幂等：同一单据多次收到"审批中/已通过"不能产生重复状态更新。

4. 解耦与扩展点

• 业务类型路由：businessMark 是跨服务交互的主键，业务方按它注册监听器，流程引擎服务按它路由消息，互不感知对方细节；
• 两级事件：引擎事件（Flowable）→ 应用内 Spring 事件 → RocketMQ 消息，每一级都可以替换或扩展；
• SDK 化：客户端 SDK 把远程调用、消息消费、监听器路由全部封装好，业务服务只需关注"我的业务在流程什么阶段要做什么"。

参考资料

• Flowable 事件与监听器官方文档
• Flowable Spring Boot 集成