如何通过 REST 异步向 Activiti 流程发送信号

结论

Activiti 可以通过 REST 唤醒处于等待状态的流程，但它不是消息代理，不会像 JMS、Kafka 或 RabbitMQ 那样订阅 channel，也不能根据 orderId 自动完成消息关联。

对于 paymentReceived 这类只对应某个订单的事件，比较合适的处理方式是：

1. 在 BPMN 中使用 message catch event、receive task 等等待状态；
2. 将 orderId 保存为流程变量，最好同时将其设为 businessKey；
3. 在 Activiti 前面增加消息适配器；
4. 由适配器消费付款事件，并根据 orderId 查询对应的 execution；
5. 通过 Activiti REST API 将 message 或 signal 投递给该 execution。

这种设计仍然是松耦合的，只是消息传输和 correlation 由集成层负责，而不是全部交给 Activiti。

为什么不能直接把 orderId 发布给 Activiti

这里涉及两个不同的概念。

Signal 是广播语义

BPMN signal 表示全局可见的事件。同名 signal 可能唤醒多个正在等待它的 execution。

如果所有订单都在等待同一个 paymentReceived signal，直接广播该 signal 可能导致多个订单流程继续执行，不符合“一笔付款只对应一个订单”的业务语义。

Activiti 的 Java API 通常也允许将 signal 定向发送给指定 execution：

runtimeService.signalEventReceived("paymentReceived", executionId);

不过，调用方仍然需要先取得 executionId。

Message 是定向语义，但仍然需要关联

BPMN message 更接近“将消息交给某个特定流程实例”。典型的 Java API 调用形式是：

runtimeService.messageEventReceived(
    "paymentReceived",
    executionId,
    variables
);

这里同样需要 executionId。Activiti 不会仅根据 message payload 中的 orderId 自动选择流程实例。

Activiti 负责维护流程状态和事件订阅。至于如何将外部业务标识映射到具体的流程等待点，系统仍然需要明确的 correlation 规则。

推荐的流程建模方式

订单流程可以设计成下面的结构：

创建订单
   |
等待 paymentReceived message
   |---------------- boundary timer ----------------> 取消订单
   |
确认付款

创建流程实例时，将订单号同时保存为 businessKey 和流程变量：

{
  "processDefinitionKey": "orderProcess",
  "businessKey": "ORDER-2026-000123",
  "variables": [
    {
      "name": "orderId",
      "value": "ORDER-2026-000123"
    }
  ]
}

businessKey 适合表示流程实例对应的业务实体。流程变量则可以用于条件判断、查询和记录其他业务数据。

需要注意，设置 businessKey 不代表 Activiti 会自动完成 message correlation。它只是提供了一个稳定且可查询的关联键。

通过 REST 完成关联和投递

不同 Activiti 版本及其 REST 应用的 endpoint 和请求体可能不同。下面展示通用的处理过程，实施时应以当前版本的 REST API 文档为准。

第一步：查询订单对应的流程实例

概念上需要执行：

GET /runtime/process-instances?businessKey=ORDER-2026-000123

也可以根据流程变量查询：

GET /runtime/process-instances?variables=orderId_eq_ORDER-2026-000123

不同版本使用的变量过滤参数格式不完全相同。

如果业务规则要求每个订单只能有一个活动流程实例，就要确认查询结果恰好为一个。返回零个或多个结果时，都不应直接投递消息。

第二步：查询正在等待消息的 execution

取得 processInstanceId 后，查询该实例中的活动 execution：

GET /runtime/executions?processInstanceId=PROCESS_INSTANCE_ID

接着确定哪个 execution 正在等待 paymentReceived。如果流程模型能保证每个订单只有一个明确的付款等待节点，也可以结合 activityId 缩小查询范围。

第三步：向 execution 投递消息

Activiti REST 常见的操作方式是更新指定 execution，并在请求体中声明操作类型。示意请求如下：

PUT /runtime/executions/EXECUTION_ID
Content-Type: application/json

{
  "action": "messageEventReceived",
  "messageName": "paymentReceived",
  "variables": [
    {
      "name": "paymentId",
      "value": "PAY-987654"
    },
    {
      "name": "paidAmount",
      "value": 199.00
    }
  ]
}

如果 BPMN 模型使用 signal catch event，对应请求通常类似：

{
  "action": "signalEventReceived",
  "signalName": "paymentReceived",
  "variables": [
    {
      "name": "paymentId",
      "value": "PAY-987654"
    }
  ]
}

具体的 action 名称和 payload 结构需要根据实际使用的 Activiti 版本确认。某些版本可能只开放部分 runtime 操作，也可能采用不同的 endpoint。

消息适配器示例

消息适配器可以订阅 Kafka、JMS 或 RabbitMQ。收到付款事件后，由它通过 REST 查询并唤醒流程。

下面是简化的 Java 伪代码：

public void onPaymentReceived(PaymentReceived event) {
    String processInstanceId =
        activitiClient.findActiveProcessInstanceByBusinessKey(event.orderId());

    if (processInstanceId == null) {
        storeAsUnmatchedEvent(event);
        return;
    }

    String executionId =
        activitiClient.findWaitingExecution(
            processInstanceId,
            "paymentReceived"
        );

    if (executionId == null) {
        handleDuplicateOrLatePayment(event);
        return;
    }

    Map<String, Object> variables = Map.of(
        "paymentId", event.paymentId(),
        "paidAmount", event.amount()
    );

    activitiClient.deliverMessage(
        executionId,
        "paymentReceived",
        variables
    );
}

调用链如下：

Payment Service
      |
      | publish PaymentReceived(orderId, paymentId)
      v
Message Broker
      |
      v
Activiti Adapter
      |
      | correlate by orderId/businessKey
      | call Activiti REST
      v
Activiti Process

Payment Service 不需要了解 Activiti 的 processInstanceId、executionId 或 BPMN 节点，所以它仍然与流程引擎解耦。只有适配器需要了解 Activiti 的 REST 协议。

更稳妥的关联方式

每次收到消息后都通过流程变量搜索数据库，数据量较大时可能带来性能和一致性问题。实践中通常可以采用以下几种方案。

使用 businessKey

直接将 orderId 设为 businessKey，查询逻辑会更清楚：

businessKey = orderId

如果同一个订单可以启动多个流程，还需要增加流程定义 key、租户或状态等限定条件。

保存关联表

启动流程时维护一张映射表：

orderId -> processInstanceId

流程进入付款等待节点后，还可以进一步登记：

orderId -> executionId

这种方式查询较快，但需要处理 execution 变化、流程结束和事务一致性。

让适配器调用自定义 correlation API

可以在 Activiti 外围提供一个面向业务的接口：

POST /process-messages

{
  "messageName": "paymentReceived",
  "correlationKey": "ORDER-2026-000123",
  "variables": {
    "paymentId": "PAY-987654",
    "paidAmount": 199.00
  }
}

接口内部负责查询订阅、校验结果唯一性，再调用 Activiti。其他服务只需要依赖稳定的业务协议，不必依赖 Activiti 的 execution REST API。

异步不等于 REST 调用本身异步

通过 REST 调用 Activiti 时，HTTP 请求本身通常仍是同步的。调用方需要等待 Activiti 接收并处理命令。

真正的异步边界来自消息代理：

生产者发布事件并返回
        ↓
消息代理持久化事件
        ↓
适配器稍后消费并调用 Activiti

如果 Payment Service 直接调用 Activiti REST，两者虽然通过网络接口通信，但仍然存在运行时耦合。Activiti 不可用时，付款处理方必须重试或报错。

必须处理的可靠性问题

重复消息

消息代理通常只能保证至少一次投递，因此同一个 paymentId 可能被多次消费。适配器必须进行幂等处理，例如为 paymentId 建立唯一约束。

如果流程已经离开付款等待节点，后续投递不应再次推进流程。

超时与付款同时发生

付款事件和 boundary timer 可能几乎同时触发。哪个分支最终成功，取决于流程引擎的并发控制和事务提交顺序。

适配器应将下面的情况视为正常竞争：

• 查询时 execution 仍然存在；
• 投递时 timer 已经先完成；
• Activiti 返回 execution 不存在或不再订阅该事件。

此时不应无限重试。适配器需要检查流程的最终状态，并将迟到的付款交给退款或人工处理逻辑。

消息早于等待状态到达

付款事件可能在流程进入 message catch event 之前到达。Activiti 的 BPMN message catch event 通常不会像消息队列一样自动保存任意提前到达的消息。

可以采用以下方式之一：

• 由消息代理重试；
• 将未匹配事件保存到 inbox 表；
• 流程进入等待节点时主动检查是否已有付款；
• 使用事务型 inbox/outbox 模式。

查询结果必须唯一

不能简单取查询结果中的第一条。应当明确处理以下情况：

0 个匹配：消息过早、流程不存在或已经结束
1 个匹配：正常投递
多个匹配：关联键设计错误或业务数据异常

REST 鉴权与重试

适配器还需要处理：

• 身份认证和权限控制；
• HTTP 超时；
• 区分 4xx 与 5xx；
• 指数退避重试；
• 死信队列；
• correlation ID 和审计日志。

可选方案

如果不想自行维护适配器，可以使用具备 Activiti 集成能力的框架或中间层，例如：

• Spring Integration；
• Apache Camel；
• JMS message-driven bean；
• Kafka/RabbitMQ consumer；
• 自定义 Activiti endpoint 或 engine plugin；
• 提供 correlation 能力的流程编排平台。

这些方案不会消除 correlation，只是将相关代码封装到可复用的集成组件中。

对于这个订单场景，较为清晰的方案是使用 BPMN message 表达定向业务事件，将 orderId 设为 businessKey，由消息适配器完成 correlation，再通过 REST 将消息投递到唯一的等待 execution。只有确实需要向多个流程订阅者广播时，才适合使用 Signal。

备注：内容仅供参考。