关于jbpm4流程并发的一点疑惑

最近由于工作上的需要，想用jbpm4实现自动节点的并发运行，首先考虑的就是fork和join节点，但是并没有达到我想看到的线程级别的并发。

先说一下的我运行的例子吧！

实验思路：设置fork的属性为 <on continue="exclusive" event="end"/>，这样就会对下面的节点生成异步job，让jobexecutor完成。

流程定义：

 

 

<?xml version="1.0" encoding="UTF-8"?>

 

<process name="AsyncFork" xmlns="http://jbpm.org/4.0/jpdl">

 

    <start g="22,69,80,40">

       <transition to="fork"/>

    </start>

 

    <fork g="99,68,80,40" name="fork">

      <on continue="exclusive" event="end"/>

    <transition g="122,41:" to="ship goods"/>

    <transition g="123,142:" to="send bill"/>

    </fork>

 

    <java class="org.jbpm.examples.async.fork.Application" g="159,17,98,50" method="shipGoods" name="ship goods">

       <transition g="297,42:" to="join"/>

    </java>

 

  <java class="org.jbpm.examples.async.fork.Application" g="159,117,98,50" method="sendBill" name="send bill">

    <transition g="297,141:" to="join"/>

  </java>

 

  <join g="274,66,80,40" name="join">

      <transition name="to java1" to="java1" g="-53,-17"/>

  </join>

 

  <end g="498,72,48,48" name="end"/>

   <java name="java1" g="356,69,92,52" class="org.jbpm.examples.async.fork.Application" method="sendApple">

      <transition name="to end" to="end" g="-41,-17"/>

   </java>

</process>

Application：输出一条打印语句，然后等待5s，模拟并发的任务

public class Application implements Serializable {

 

  private static final long serialVersionUID = 1L;

 

     public void shipGoods() {

       // assume automatic calculations here

         try {

            System.out.println("正在输出PDF...... ");

              Thread.sleep(4000);

              System.out.println("输出PDF完毕！！");

           } catch (InterruptedException e) {

              // TODO Auto-generated catch block

              e.printStackTrace();

           }

     }

 

     public void sendBill() {

       // assume automatic calculations here

      try {

                System.out.println("正在输出Primes...... ");

              Thread.sleep(6000);

              System.out.println("输出Primes完毕！！");

           } catch (InterruptedException e) {

              // TODO Auto-generated catch block

              e.printStackTrace();

           }

     }

     public void sendApple() {

        // assume automatic calculations here

         try {

                   System.out.println("正在输出Apple...... ");

                  Thread.sleep(5000);

                  System.out.println("输出Apples完毕！！");

              } catch (InterruptedException e) {

                  // TODO Auto-generated catch block

                  e.printStackTrace();

              }

      }

}

运行

1.       在 jbpm.cfg.xml 添加  <import resource="jbpm.jobexecutor.cfg.xml" />

2.添加文件jbpm.jobexecutor.cfg.xml。内容如下：

<?xml version="1.0" encoding="UTF-8"?> 

<jbpm-configuration> 

<process-engine-context> 

<job-executor threads="2" idle="15000" idle-max="20000" lock-millis="25000" /> 

</process-engine-context> 

</jbpm-configuration> 

3.部署流程到数据库

4.运行流程。

public static void main(String[] args) {

       // TODO Auto-generated method stub

       ProcessEngine processEngine = null;

       ExecutionService executionService = null;

       ManagementService managementService=null;

      

       Configuration configuration= new Configuration();

       processEngine = configuration.buildProcessEngine();

      

       executionService = processEngine.getExecutionService();

       managementService=processEngine.getManagementService();

      

        ProcessInstance processInstance=

            executionService.startProcessInstanceByKey("AsyncFork");

    }

效果如下：

 

11:35:51,578 FIN   | [ExecuteActivity] executing activity(ship goods)

正在输出PDF......

11:35:51,609 FIN     | [ExecuteJobCmd] executing job message[103]...

正在输出Primes......

11:35:51,625 FIN     | [ExecuteActivity] executing activity(send bill)

输出PDF完毕！！

11:35:55,578 FIN   | [ExecuteActivity] executing activity(join)

输出Primes完毕！！

11:35:57,625 FIN     | [ExecuteActivity] executing activity(join)

………

问题是：并没有输出：正在输出Apple......  和 输出Apples完毕！！也就是说流程推进到join节点那，就进行不下去了。

原因：

我断点调试了一下，原来是异步环境下，线程A对节点状态做的修改不能同步到线程B中的。而join是当两个节点的状态都是inactive-join时，才能继续向下推进。

假如线程A对应节点1，线程B对应node2,两个线程同时启动。A先修改node1的状态为inactive-join，然后join会判断是否都完成。然后线程B修改node2的状态为inactive-join，但是在接着join判断都完成时，由于node1的状态并没有同步过来，导致join节点完成继续向下推进。

求助

大家看看是我代码或配置的问题，还是别的方面的原因，希望大家给点建议。