Camunda并行任务的处理

在工作流的编排中，有些时候会同时运行多个任务或子进程，默认的方式是串行运行，但是为了提高性能，我们会希望能并行运行。我也进行了一些测试，发现这个并行运行还不是这么简单的。

举个例子，我们现在定义一个流程，这个流程很简单，就是调用一个Http接口。然后再定义另外一个流程，在这个流程里面引用刚才定义的子流程，然后运行这个子流程多次。我们希望实现的效果是这个子流程的运行能够并行。

首先我们先简单定义一个Http接口，在这个接口接收到Reqest之后，简单的休眠10秒之后再返回。这个接口很简单，代码就不附上了。

然后我们定义一个简单的流程，里面包含一个Service Task，是调用这个Http接口的。这个流程的BPMN内容如下：

<?xml version="1.0" encoding="UTF-8"?>
<bpmn2:definitions xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:bpmn2="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" xmlns:camunda="http://camunda.org/schema/1.0/bpmn" id="sample-diagram" targetNamespace="http://bpmn.io/schema/bpmn" xsi:schemaLocation="http://www.omg.org/spec/BPMN/20100524/MODEL BPMN20.xsd">
  <bpmn2:process id="Process_12345" name="Test_12345" isExecutable="true">
    <bpmn2:startEvent id="StartEvent_1">
      <bpmn2:outgoing>Flow_1lqf9ii</bpmn2:outgoing>
    </bpmn2:startEvent>
    <bpmn2:sequenceFlow id="Flow_1lqf9ii" sourceRef="StartEvent_1" targetRef="Activity_1jsutqa" />
    <bpmn2:endEvent id="Event_03vs1kv" camunda:asyncBefore="true">
      <bpmn2:incoming>Flow_179keqc</bpmn2:incoming>
    </bpmn2:endEvent>
    <bpmn2:sequenceFlow id="Flow_179keqc" sourceRef="Activity_1jsutqa" targetRef="Event_03vs1kv" />
    <bpmn2:serviceTask id="Activity_1jsutqa" name="调用HTTP接口" camunda:resultVariable="newNEType">
      <bpmn2:extensionElements>
        <camunda:connector>
          <camunda:inputOutput>
            <camunda:inputParameter name="url">http://127.0.0.1:7777/getDeviceList</camunda:inputParameter>
            <camunda:inputParameter name="method">POST</camunda:inputParameter>
            <camunda:inputParameter name="headers">
              <camunda:map>
                <camunda:entry key="Accept">application/json</camunda:entry>
                <camunda:entry key="Content-Type">application/json</camunda:entry>
              </camunda:map>
            </camunda:inputParameter>
            <camunda:inputParameter name="payload">{"neType": "${execution.getVariable("neType")}",
"test": "${execution.getVariable("test")}"}</camunda:inputParameter>
          </camunda:inputOutput>
          <camunda:connectorId>http-connector</camunda:connectorId>
        </camunda:connector>
      </bpmn2:extensionElements>
      <bpmn2:incoming>Flow_1lqf9ii</bpmn2:incoming>
      <bpmn2:outgoing>Flow_179keqc</bpmn2:outgoing>
    </bpmn2:serviceTask>
  </bpmn2:process>
  <bpmndi:BPMNDiagram id="BPMNDiagram_1">
    <bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Process_12345">
      <bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
        <dc:Bounds x="192" y="102" width="36" height="36" />
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape id="Event_03vs1kv_di" bpmnElement="Event_03vs1kv">
        <dc:Bounds x="432" y="102" width="36" height="36" />
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape id="Activity_1fzq8lj_di" bpmnElement="Activity_1jsutqa">
        <dc:Bounds x="280" y="80" width="100" height="80" />
        <bpmndi:BPMNLabel />
      </bpmndi:BPMNShape>
      <bpmndi:BPMNEdge id="Flow_1lqf9ii_di" bpmnElement="Flow_1lqf9ii">
        <di:waypoint x="228" y="120" />
        <di:waypoint x="280" y="120" />
      </bpmndi:BPMNEdge>
      <bpmndi:BPMNEdge id="Flow_179keqc_di" bpmnElement="Flow_179keqc">
        <di:waypoint x="380" y="120" />
        <di:waypoint x="432" y="120" />
      </bpmndi:BPMNEdge>
    </bpmndi:BPMNPlane>
  </bpmndi:BPMNDiagram>
</bpmn2:definitions>

然后我们再定义另外一个流程，这个流程里面包含一个Call Activity，这个Call Activity引用了刚才我们定义的流程。为了能让这个Call Activity能运行多次，我们需要定义一个inputCollection的变量，例如我在工作流里面定义一个Script Task，然后通过JavaScript来创建一个InputCollection：

var ArrayList = Java.type("java.util.ArrayList")
var neTypes = new ArrayList();
neTypes.add('AMF');
neTypes.add('SMF');
neTypes.add('UPF');
neTypes.add('ENUM');
neTypes.add('SPG');
neTypes.add('PCC');
neTypes;

然后创建一个Call Activity，在配置里面选择类型是并行运行多实例，即三条竖线的小图标，然后在Multi-Instance属性里面的Collection，设置为${neTypes}，即刚才Script Task输出的变量。在Element Variable里面输入neType，这样当工作流运行时，就会遍历Collection里面的元素，把其赋值为neType，然后传给Call Activity执行。另外，在Multi-Instance里面的Asynchronous before选项也要勾上，但是Exclusive选项不要勾选。这样当流程执行到Call Activity之前，会暂停，然后创建线程任务来异步运行Call Activity。如果不这样设置，那么即使我们设置Call Activity是并行运行，但是仍然是串行执行。如下图所示：

 以下是这个流程定义的BPMN代码：

<?xml version="1.0" encoding="UTF-8"?>
<bpmn2:definitions xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:bpmn2="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" xmlns:camunda="http://camunda.org/schema/1.0/bpmn" id="sample-diagram" targetNamespace="http://bpmn.io/schema/bpmn" xsi:schemaLocation="http://www.omg.org/spec/BPMN/20100524/MODEL BPMN20.xsd">
  <bpmn2:process id="Process_12345" name="Test_12345" isExecutable="true">
    <bpmn2:startEvent id="StartEvent_1">
      <bpmn2:outgoing>Flow_1lqf9ii</bpmn2:outgoing>
    </bpmn2:startEvent>
    <bpmn2:sequenceFlow id="Flow_1lqf9ii" sourceRef="StartEvent_1" targetRef="Activity_1jsutqa" />
    <bpmn2:endEvent id="Event_03vs1kv" camunda:asyncBefore="true">
      <bpmn2:incoming>Flow_179keqc</bpmn2:incoming>
    </bpmn2:endEvent>
    <bpmn2:sequenceFlow id="Flow_179keqc" sourceRef="Activity_1jsutqa" targetRef="Event_03vs1kv" />
    <bpmn2:serviceTask id="Activity_1jsutqa" name="调用HTTP接口" camunda:resultVariable="newNEType">
      <bpmn2:extensionElements>
        <camunda:connector>
          <camunda:inputOutput>
            <camunda:inputParameter name="url">http://127.0.0.1:7777/getDeviceList</camunda:inputParameter>
            <camunda:inputParameter name="method">POST</camunda:inputParameter>
            <camunda:inputParameter name="headers">
              <camunda:map>
                <camunda:entry key="Accept">application/json</camunda:entry>
                <camunda:entry key="Content-Type">application/json</camunda:entry>
              </camunda:map>
            </camunda:inputParameter>
            <camunda:inputParameter name="payload">{"neType": "${execution.getVariable("neType")}",
"test": "${execution.getVariable("test")}"}</camunda:inputParameter>
          </camunda:inputOutput>
          <camunda:connectorId>http-connector</camunda:connectorId>
        </camunda:connector>
      </bpmn2:extensionElements>
      <bpmn2:incoming>Flow_1lqf9ii</bpmn2:incoming>
      <bpmn2:outgoing>Flow_179keqc</bpmn2:outgoing>
    </bpmn2:serviceTask>
  </bpmn2:process>
  <bpmndi:BPMNDiagram id="BPMNDiagram_1">
    <bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Process_12345">
      <bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
        <dc:Bounds x="192" y="102" width="36" height="36" />
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape id="Event_03vs1kv_di" bpmnElement="Event_03vs1kv">
        <dc:Bounds x="432" y="102" width="36" height="36" />
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape id="Activity_1fzq8lj_di" bpmnElement="Activity_1jsutqa">
        <dc:Bounds x="280" y="80" width="100" height="80" />
        <bpmndi:BPMNLabel />
      </bpmndi:BPMNShape>
      <bpmndi:BPMNEdge id="Flow_1lqf9ii_di" bpmnElement="Flow_1lqf9ii">
        <di:waypoint x="228" y="120" />
        <di:waypoint x="280" y="120" />
      </bpmndi:BPMNEdge>
      <bpmndi:BPMNEdge id="Flow_179keqc_di" bpmnElement="Flow_179keqc">
        <di:waypoint x="380" y="120" />
        <di:waypoint x="432" y="120" />
      </bpmndi:BPMNEdge>
    </bpmndi:BPMNPlane>
  </bpmndi:BPMNDiagram>
</bpmn2:definitions>

还有一点需要注意的，那就是在Call Activity引用的流程里面，在end event那里也需要设置Asynchronous before和Exclusive，不然的话Call Activity运行的次数会和我们预计的不同，会更多一些。按照网上的解释，由于可能存在多个线程同时运行这个子进程，但是Camunda只允许同一个进程中只有一个线程是激活的，因此只有1个线程是成功运行，其他的线程将重试运行，因此我们就会观察到有多于我们预计的运行次数。在End envent设置Async before可以阻止线程重复运行。

运行流程，从HTTP接口打印的调用日志，可以看到子流程确实是并行运行的。

在实际项目中，我还发现了一个奇怪的问题，就是我在Call Activity并行运行多个子进程，但是其中一个子进程会一直运行而不结束，其他子进程则没有这个问题。当时想了好久也没找到有啥问题，后来观察了一下这个子进程，其运行时间较长，因为要处理的数据比其他子进程要多，大概运行一次要5分多钟的时间。我又查看了一下官网上的配置描述，发现其中在任务执行里面有一个配置参数是lock-time-in-millis，默认配置刚好是5分钟，因此很有可能是因为超过5分钟这个任务还没结束，因此Camunda又启动了另外一个任务来重复执行，因此造成的现象就是好像这个任务一直在运行。找到问题之后，我把这个配置改为15分钟，即解决问题。当然按照官网的解释，另一种更好的解决办法是，当有任务需要运行长时间时，最好是改为external task的方式交由外部执行，当外部执行完毕之后发送消息给任务即可。