Activiti 并行服务任务

Activiti parallel Service Tasks

我正在尝试在 Activiti 中实现两个服务任务，它们应该 运行 并行。下面编写的代码可以随机（有趣地）正常工作。

我的意思是它偶尔只打印“first”（或“second”）或者打印两个“first”一个“[=13=” ]”等

问题：如何使这些服务 运行 不断并行；不管当前有多少服务 运行ning?

PS：当我从流程定义中删除 activiti:async="true" 时，它只打印“first”或“second”。我想我需要那个:)

进程定义

<?xml version='1.0' encoding='UTF-8'?>
<definitions xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL"
xmlns:activiti="http://activiti.org/bpmn" targetNamespace="Examples">

    <process id='testparallelact' name="Developer Hiring" isExecutable="true" activiti:exclusive="false" activiti:async="true">

        <startEvent id="theStart" />
        <sequenceFlow id="flow1" sourceRef="theStart" targetRef="fork" />

        <parallelGateway id="fork"  activiti:async="true" />
        <sequenceFlow sourceRef="fork" targetRef="receivePayment" />
        <sequenceFlow sourceRef="fork" targetRef="shipOrder" />


        <serviceTask id="receivePayment" name="Receive Payment" activiti:async="true" activiti:exclusive="false"
activiti:expression="${serviceConnections.runThis2('First')}"/>

        <sequenceFlow sourceRef="receivePayment" targetRef="join" />


        <serviceTask id="shipOrder" name="Ship Order" activiti:async="true" activiti:exclusive="false"
activiti:expression="${serviceConnections.runThis2('Second')}"/>

        <sequenceFlow sourceRef="shipOrder" targetRef="join" />

        <parallelGateway id="join" />
        <sequenceFlow sourceRef="join" targetRef="theEnd" />
        <endEvent id="theEnd" />
    </process>
</definitions>

流程定义的图形呈现

"runThis2"

代码 
public void runThis2(String test1) throws InterruptedException {            
    while(true)
    {
        Thread.sleep(1000);
        System.out.println(test1);              
    }           
}

"async" 和 "exclusive" 标志的组合很重要

了解作业在 Activiti 引擎中的执行方式很重要。 下面的论坛帖子很好地描述了它：

https://community.alfresco.com/thread/221453-multiinstance-wont-run-task-in-parallel

Activiti 最初的架构师之一 Tijs Rademakers 在 2013 年 10 月 25 日的重要摘录：

The parallel gateway and multiinstance constructs are able to run multiple user tasks in parallel for example. But for service and script tasks they are basically executed serial still. Async can change this behavior if you also set exclusive to false (the default is true). Then the job executor will just execute all jobs available and not serially. So give it a try to set async to true and exclusive to false.

现在，通过设置 activiti:async="true" 和 activiti:exclusive="false"，您有效地完成了通过将服务任务（通常是串行处理的）分配给作业来在进程中创建一个 "wait state"执行者.

但是：

  • 并行执行了多少个作业并且
  • 当它们被作业调度程序执行时

现在完全由 Job Executor 配置控制。 （线程池的大小、超时、并发作业的数量、作业块的大小都是可配置的。）

现在，这并不完全是您所期望的，这意味着，它取决于您的作业队列的大小、单次扫描中执行的作业数量以及每个作业的持续时间以及服务任务何时完成被执行。意思是，它们可能并行执行，也可能串行执行。同样，您无法控制它们的顺序，因为再次由作业执行器决定它做什么和什么时候。

好的，假设这符合您的要求...

乐观锁概念

...还有一个问题您可能会遇到（事实上，这是首先引入 activiti:exclusive 标志的原因）。服务任务完成后，执行上下文将提交到数据库中的流程实例记录以及历史记录。出于性能目的，Activiti 使用 "Optimistic Locking" 作为记录。

现在，如果您的流程分支在时间上相对接近地完成，那么您可能（实际上很有可能）会收到关于数据库更新的 Optimistic Locking Exception，如下所示：

09:59:52,432 [flowable-async-job-executor-thread-2] ERROR org.flowable.job.service.impl.asyncexecutor.DefaultAsyncRunnableExecutionExceptionHandler - Job 12575 failed org.flowable.engine.common.api.FlowableOptimisticLockingException: ProcessInstance[12567] was updated by another transaction concurrently

（注意：上面的错误实际上不是来自 Activiti，而是来自一个名为“Flowable”的项目。然而，它们与当时的 Activiti 6 具有基本相同的代码库最初提出问题。（2017 年 11 月）。）

这将导致服务任务被标记为失败，并且将重新尝试。如果您对 SOR（记录系统）或其他遗留系统进行外部调用，这可能会出现问题。 （考虑一下如果您的航班实际上已成功预订，但对 reserve 的调用是第二次，因为它被认为失败了。）

所有有趣的事情以及所有可以通过良好的设计和使用最佳实践来解决的事情。

希望这可以帮助您了解正在发生的事情。

Greg@BP3

进一步阅读

Alfresco 论坛 post 包含几个死的 link。下面是直播 links.

死问题跟踪器 links

补充说明：Activiti 目前（2019 年）不再使用这两个（codehaus.org 或 atlassian.net）跟踪器。相反，他们使用这个 GitHub 跟踪器：https://github.com/Activiti/Activiti/issues

死亡常见问题link

Activiti 标志

Camunda 手册