Flowable-服务-骆驼任务

定义

Camel 任务不是 BPMN 2.0 规范定义的官方任务，在 Flowable 中，Camel 任务是作为一种特殊的服务
任务来实现的。主要做路由工作的。

图形标记

由于 Camel 任务不是 BPMN 2.0 规范的“官方”任务，因此没有提供其专用图标。在 bpmnjs
中 Camel 任务我们就使用服务任务代替，如图所示：

在这里插入图片描述

XML内容

Camel 任务在 Flowable 中定义为一个专用的服务任务，这个服务任务的 type 属性设置为 camel。
Camel 任务的 XML 表示格式如下：

<serviceTask id="camelTask1" name="Camel " flowable:type="camel" />

以上在流程定义的服务任务上定义 Camel 类型即可，集成逻辑都将通过 Camel 容器委托。

Flowable与Camel集成

本节将具体介绍 Flowable 与 Camel 集成的过程，以及 Flowable 基于 Camel 扩展出来的各种特性
和用法。

Camel 的配置与依赖
默认情况下，使用 Camel 任务时，Flowable 引擎在 Spring 容器中查找 camelContext Bean。camelContext
Bean 用于定义 Camel 容器装载的 Camel 路由，Camel 路由可以在 Spring 配置文件中定义，也可以按照指
定的 Java 包装载路由：

   <camelContext id="camelContext" xmlns="http://camel.apache.org/schema/spring">
    <packageScan>
        <package>com.dragon.test.bpmn.camel.route</package>
    </packageScan>
   </camelContext>

通过以上配置，在初始化时 CamelContext 时会把 com.dragon.test.bpmn.camel.route 中
的路由定义类（继承自 RouteBuilder，后文会讲解）注册到 CamelContext 对象中。CamelContext 是 Camel
中一个很重要的概念，它横跨了 Camel 服务的整个生命周期，并且为 Camel 服务的工作环境提供支撑，
Camel 中的各个 Service 的关联衔接通过 CamelContext 上下文对象完成。
由于 Flowable 的配置文件采用的是 spring bean 配置文件格式，因此在 Flowable 与 Camel 集成时，以上
配置内容可以直接加在 Flowable 的配置文件中。
如果想要定义多个 camelContext，或想使用不同的 Bean 名字，可以在 Camel 任务定义中通过以下方
式指定：

<serviceTask id="camelTask1" name="Camel " flowable:type="camel" />

需要注意的是，如果要使用 Camel 任务，需要在项目中包含 flowable-camel 模块依赖及 Camel 相关依
赖，Maven 依赖定义如下：

   <dependency>
       <groupId>org.flowable</groupId>
       <artifactId>flowable-camel</artifactId>
       <version>xxx</version>
   </dependency>
   <dependency>
       <groupId>org.apache.camel</groupId>
       <artifactId>camel-http</artifactId>
       <version>xxx</version>
   </dependency>

定义Camel路由
Camel 最重要的特色之一就是路由，路由用于应用中的通信或者应用间通信。Camel 的路由需要通过
我们手动编排的方式，在指定的（或可变的）端点间进行数据的传输、过滤、转换等操作。Camel 路由
易于使用的一个特性是端点 URI，通过指定 URI，可以确定要使用的组件以及该组件是如何被配置的。
然后可以决定将消息发送到由该 URI 配置的组件，或者使用该组件发出消息。
flowable 的 flowable-camel 模块提供了 Camel 与 Flowable 通信的桥梁，当流程流转到 Camel 任务后流程
引擎将调用 Camel 执行对应的路由，同时可以选择把流程变量传递给路由，在路由处理结束后还可以有
选择地把路由得到的结果以流程变量的方式回传给流程实例。
我们可以通过 Java DSL 构建 Camel 路由（也可通过 XML 配置，这里不做介绍），需要继承
org.apache.camel.builder.RouteBuilder 类，然后实现其中的 configure 方法。Flowable 与 Camel 集成后，一个
典型的路由定义类如下：

   public class GetIpInfoCamelCallRoute extends RouteBuilder {
    @Override
    public void configure() {
        from("flowable:CamelTaskProcessTest:afef07cc915df4771a54048fe871e9c9e?copyVariablesToProperties=true")
                .toD("http://ip-api.com/json/${exchange.getProperty('ip')}?lang=zh-CN&bridgeEndpoint=true")
                .process(new ResultProcessor());
    }
}

在以上路由定义类的 configure 方法中，通过 Java 的 DSL 语言（域特定语言）描述路由规则，from
和 to 是两个关键字，Camel 会从 from 声明的起始端点将消息路由至 to 声明的终点。所有的路由都以一
个 from 开始，它接受端点 URI 作为参数。flowable-camel 模块定义了“flowable”类型的路由 URI 协议，以
上面的配置为例，from 端点（endpoint）的格式包含冒号和问号分隔的几个部分，各部分的含义如表 15.3
所示。
flowable-camel 模块提供的 URI 协议

属性	描述
flowable	协议开头，指向引擎端点
CamelTaskProcessTest	流程定义 Key
afef07cc915df4771a54048fe871e9c9e	流程定义中 Camel 服务的 id
copyVariablesToProperties=true	路由 URI 的参数

路由URI配置
我们可以通过在 URI 中附加一些参数，使用 Flowable 提供的几种不同的行为，从而实现干预 Camel
组件的功能。本节将介绍 URI 中支持的参数，包括输入参数和输出参数两类。
3.1. 输入参数
Flowable 提供了三种输入参数，可以用于控制将流程变量复制到 Camel 的策略，可配置的输入参数如
表所示：

参数	对应flowable行为类	描述
copyVariablesToProperties	org.flowable.camel.impl.CamelBehaviorDefaultImpl	默认配置，将 Flowable 的流程变量复制为 Camel 参数，在路由中可以通过形如${property.variableName}的表达式获取到参数值。
copyCamelBodyToBody	org.flowable.camel.impl.CamelBehaviorCamelBodyImpl	只将名为"camelBody"的 Flowable 流程变量复制为 Camel 消息体。如果 camelBody 的值是 Map 对象，在路由中可以通过形如 ${body[variableName]}的表达式获取到参数值；如果 camelBody 的值是纯字符，可以使用$ {body}表达式获取。
copyVariablesToBodyAsMap	org.flowable.camel.impl.CamelBehaviorBodyAsMapImpl	把 Flowable 的所有流程变量复制到一个 Map 对象里，作为 Camel 的消息体。，在路由中可以通过形如${body[variableName]}的表达式获取到参数值。
以如下路由规则为例：

   from("activiti:CamelTaskProcess:camelTask1?copyVariablesToProperties=true")
   .to("log:org.activiti.camel.examples.SimpleCamelCall");

这里的配置在 URI 中附加了 copyVariablesToProperties=true，表示将 Flowable 的流程变量复制成 Camel
参数。
3.2. 输出参数
同样的，Flowable 提供了几种输出参数，可以用于控制将 Camel 执行结果复制到流程变量的策略，可
配置的输出参数如表所示：
flowable-camel 模块提供的 URI 协议：

参数	描述
default	默认配置。如果 Camel 消息体是一个 Map 对象，则在路由执行结束后将其中每一个属性复制为 Flowable 的流程变量；否则将整个 Camel 消息体复制到名为"camelBody"的流程变量。
copyVariablesFromProperties	将 Camel 参数以相同的名称复制为 Flowable 流程变量。
copyVariablesFromHeader	将 Camel Header 中的内容以相同的名称复制为 Flowable 流程变量。
copyCamelBodyToBodyAsString	与 default 相同，但如果 Camel 消息体不是 Map 对象，则首先将其转换为字符串，然后再复制到名为"camelBody"的流程变量
以如下路由规则为例：

   from("activiti:CamelTaskProcess:camelTask1?copyVariablesFromProperties=true")
   .to("log:org.activiti.camel.examples.SimpleCamelCall");

这里的配置在 URI 中附加了 copyVariablesFromProperties=true，表示将 Camel 参数以相同的名称复制
为 Flowable 流程变量中。
4. 异步Camel调用
默认情况下，Camel 任务是同步执行的，流程执行到 Camel 任务后将处于等待状态，直到 Camel 执
行结束并返回结果后，才离开 Camel 任务往下继续流转。如果 Camel 任务执行时间比较长，或者某些场
景下不需要同步执行，则可以使用 Camel 任务的异步功能，将 Camel 任务的 async 参数设置为 true 即可
启用这个功能：

   <serviceTask id="camelTask1" name=" Camel " activiti:type="camel"
   activiti:async="true"/>

设置这个参数后，Camel 路由会由 Flowable 作业执行器异步启动。
5. 通过Camel启动流程
前面几节介绍了如何整合 Camel 与 Flowable，以及两者如何通信，都是先启动 Flowable 流程实例，然
后在流程实例中启动 Camel 路由。反过来，也可以通过 Camel 任务启动或调用流程实例，其 Camel 的路
由规则可以这么设计：

   from("flowable:ParentProcess:camelTaskForStartSubprocess")
   .to("flowable:subProcessCreateByCamel");

其中，from 声明的起始端点的 URI 分为三个部分："flowable"协议开头，父流程定义 key，Camel 任务
id；to 声明的终止端点的 URI 包含两个部分："flowable"协议开头，子流程定义 key。

使用示例

下面我们看一个使用 Camel 任务的示例流程，如图所示，用于调用外部第三方服务自动化地
获取 IP 信息。流程发起后首先通过 Camel 任务调用外部 Web 服务查询 IP 信息，然后通过邮件任务发送
查询结果给申请人

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xbHKxnwk-1690689579641)(…/public/images/service/camel-task.png)]

Camel 路由配置整合到 Flowable 的配置文件中内容如下：

<process id="CamelTaskProcessTest" name="服务-骆驼任务" isExecutable="true">
    <startEvent id="a4192d69b9d5a4d1faf42b7c9741d0d48" />
    <sequenceFlow id="a0db08b7fe8d74803b11b59179c6e0a57" sourceRef="a4192d69b9d5a4d1faf42b7c9741d0d48" targetRef="afef07cc915df4771a54048fe871e9c9e" />
    <serviceTask id="afef07cc915df4771a54048fe871e9c9e" name="获取Ip信息" flowable:type="camel">
      <extensionElements>
        <flowable:formData />
        <flowable:assigneeType>static</flowable:assigneeType>
      </extensionElements>
    </serviceTask>
    <sequenceFlow id="Flow_1ce3l0r" sourceRef="afef07cc915df4771a54048fe871e9c9e" targetRef="Activity_1n9nrw6" />
    <serviceTask id="Activity_1n9nrw6" name="发送邮件" flowable:type="mail">
      <extensionElements>
        <flowable:field name="to">
          <flowable:expression>liuwenjun05101@163.com</flowable:expression>
        </flowable:field>
        <flowable:field name="subject">
          <flowable:string>IP信息查询结果</flowable:string>
        </flowable:field>
        <flowable:field name="html">
          <flowable:expression>&lt;html&gt;&lt;body&gt;&lt;p&gt;用户&amp;lt;b&amp;gt;${userName}&amp;lt;/b&amp;gt;你好，&amp;lt;br/&amp;gt;&lt;/p&gt;&lt;p&gt;你查询的IP&amp;lt;b&amp;gt;${ip}&amp;lt;/b&amp;gt;的信息为：&amp;lt;br/&amp;gt;&lt;/p&gt;&lt;p&gt;省份：${regionName} 城市：${city}&lt;/p&gt;&lt;p&gt;互联网服务提供商：${isp}&lt;/p&gt;&lt;/body&gt;&lt;/html&gt;</flowable:expression>
        </flowable:field>
        <flowable:field name="charset">
          <flowable:string>utf-8</flowable:string>
        </flowable:field>
      </extensionElements>
    </serviceTask>
    <endEvent id="Event_1l6yldo" />
    <sequenceFlow id="Flow_1h6rd9m" sourceRef="Activity_1n9nrw6" targetRef="Event_1l6yldo" />
  </process>

在以上配置中：
1、流程引擎配置使用的是 SpringProcessEngineConfiguration，这是因为 Flowable Camel集成时，
需要通过 SpringProcessEngineConfiguration 获取 camelContext，相关细节读者可自行查看源代码。
2、Camel 路由是通过 Spring 的环境下扫描路由配置实现的， Spring 会扫描包路径
com.dragon.test.bpmn.camel.route下的 Route 类加载到 camelContext 中。

设计Came路由代码

Camel 路由 Route 的代码如下：

package com.didichuxing.bpm.demo.chapter15.camel.route;
import com.didichuxing.bpm.demo.chapter15.camel.processor.ResultProcessor;
import com.didichuxing.bpm.demo.chapter15.camel.processor.TransformProcessor;
import org.apache.camel.builder.RouteBuilder;
public class GetIpInfoCamelCallRoute extends RouteBuilder {
@Override
public void configure() throws Exception {
    from("activiti:CamelTaskProcess:camelTask1?copyVariablesToProperties=true")
    .toD("http://ip-api.com/json/${property.ip}?lang=zh-CN&bridgeEndpoint=true")
    .process(new ResultProcessor());
}
}

在以上代码中：
⑴from 声明的起始端点的 URI 中，flowable 为协议开头，CamelTaskProcess 为流程定义 key，
camelTask1 为 Camel 任务 id，输入参数配置的 copyVariablesToProperties=true 表示将 Activiti 的流程
变量复制成 Camel 参数，输出参数使用的默认配置。
⑵终止端点采用 toD 声明，它允许通过表达式的方式来动态定义消息的接收节点，这里使用了
表达式${property.ip}，表示从 Camel 参数中获取 ip 属性的值。
⑶路由中使用到了自定义 Processor 处理器 ResultProcessor。Processor 处理器是 Camel 中的一
个重要元素，它用于接收从控制端点、路由选择条件又或者另一个处理器的 Exchange 中传来的消
息信息，并进行处理。我们可以在自定义的 Processor 处理器中做很多事情，比如这里使用它来进
行外部服务返回结果数据格式的转换，代码如下

import com.alibaba.fastjson.JSON;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import java.util.Map;
import java.util.stream.Collectors;
public class ResultProcessor implements Processor {
    public void process(Exchange exchange) {
        // camel
        String camelCallResult =exchange.getIn().getBody(String.class);
        // camel
        System.out.println("camel" + camelCallResult);
        // Map
        Map<String, String> camelCallResultMap = JSON.parseObject(camelCallResult,
        Map.class);
        //
        Map<String, String> resultMap =
        camelCallResultMap.entrySet().stream()
        .filter(map -> "country".equals(map.getKey()) ||
        "regionName".equals(map.getKey())
        || "city".equals(map.getKey()) ||
        "isp".equals(map.getKey()))
        .collect(Collectors.toMap(Map.Entry::getKey,
        Map.Entry::getValue));
        // Camel Map
        exchange.getOut().setBody(resultMap, Map.class);
    }
}