JBPM4学习总结之5——Fork-join（分支/聚合活动）

当我们需要流程并发处理、执行的时候就需要分支、聚合。当出现fork的时候，流程实例不是仅仅进入其中一个分支，而是所有分支都会被激活，直到外界接口执行此分支才会往下走，所有的分支节点都聚合到join的时候，整个分支才会继续往下面走，否则，最先到达join分支的节点会一直等待在join那里等待其他。

流程图如下：

 

流程描述如下：

 

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

 

<process name="forkjoin" xmlns="http://jbpm.org/4.4/jpdl">

    <start g="286,17,48,48" name="start1">

       <transition g="-52,-22" name="to fork" to="fork" />

    </start>

    <fork g="287,96,48,48" name="fork">

       <transition g="113,119:-56,-22" name="to sendEmail" to="sendEmail" />

       <transition g="314,200:0,-30" name="to sendMSN" to="sendMSN" />

       <transition g="548,120:-56,-22" name="to QQ" to="QQ" />

    </fork>

    <state g="66,175,92,52" name="sendEmail">

       <transition g="110,304:-49,-22" name="to join1" to="join1" />

    </state>

    <state g="181,173,92,52" name="sendMSN">

       <transition g="228,304:-49,-22" name="to join1" to="join1" />

    </state>

    <state g="503,225,92,52" name="QQ">

       <transition g="551,455:-49,-22" name="to join2" to="join2" />

    </state>

    <join g="143,281,48,48" name="join1">

       <transition g="-56,-22" name="to state4" to="drive" />

    </join>

    <state g="121,369,92,52" name="drive">

       <transition g="167,453:-49,-22" name="to join2" to="join2" />

    </state>

    <join g="265,432,48,48" name="join2">

       <transition g="-50,-22" name="to end1" to="end1" />

    </join>

    <end g="267,532,48,48" name="end1" />

</process>

测试代码如下：

package code;

import java.util.Set;

import junit.framework.TestCase;

import org.jbpm.api.Configuration;
import org.jbpm.api.ExecutionService;
import org.jbpm.api.ProcessEngine;
import org.jbpm.api.ProcessInstance;
import org.jbpm.api.RepositoryService;

public class Forkjoin extends TestCase {

	// 加载myjbpm.cfg.xml文件
	ProcessEngine processEngine = new Configuration().setResource(
			"myjbpm.cfg.xml").buildProcessEngine();

	// 资源服务，流程模板的部署、查询、删除

	RepositoryService repositoryService = processEngine.getRepositoryService();

	// 执行服务，设置流程实例的发起、执行
	ExecutionService executionService = processEngine.getExecutionService();

	/**
	 * 发布流程
	 */
	public void test01Deploy() {

		String deployId = repositoryService.createDeployment()
				.addResourceFromClasspath("forkjoin.jpdl.xml").deploy();

		System.out.println(deployId);
	}

	/**
	 * 发起流程，开始分开流程
	 */
	public void test02startFork() {

		// 开始process流程
		ProcessInstance processInstance = executionService
				.startProcessInstanceByKey("forkjoin");

		// 流程实例的ID
		String pid = processInstance.getId();

		System.out.println(pid);// forkjoin.10001

	}

	/**
	 * 判断有三个活动的，等待着的节点
	 */
	public void test03EqualActiveNames() {

		ProcessInstance processInstance = executionService
				.createProcessInstanceQuery().processInstanceId(
						"forkjoin.10001").uniqueResult();

		Set<String> activeNames = processInstance.findActiveActivityNames();

		System.out.println(activeNames);
	}

	/**
	 * 执行SendEmail节点
	 */
	public void test04ExeSendEmail() {

		ProcessInstance processInstance = executionService
				.createProcessInstanceQuery().processInstanceId(
						"forkjoin.10001").uniqueResult();

		String eid = processInstance.findActiveExecutionIn("sendEmail").getId();

		processInstance = executionService.signalExecutionById(eid);

		Set<String> activeNames = processInstance.findActiveActivityNames();

		System.out.println(activeNames);

	}

	/**
	 * 执行SendMSN节点
	 */
	public void test04ExeSendMSN() {

		ProcessInstance processInstance = executionService
				.createProcessInstanceQuery().processInstanceId(
						"forkjoin.10001").uniqueResult();

		String eid = processInstance.findActiveExecutionIn("sendMSN").getId();

		processInstance = executionService.signalExecutionById(eid);

		Set<String> activeNames = processInstance.findActiveActivityNames();

		System.out.println(activeNames);

	}
	
	/**
	 * 执行QQ节点
	 */
	public void test04ExeQQ() {

		ProcessInstance processInstance = executionService
				.createProcessInstanceQuery().processInstanceId(
						"forkjoin.10001").uniqueResult();

		String eid = processInstance.findActiveExecutionIn("QQ").getId();

		processInstance = executionService.signalExecutionById(eid);

		Set<String> activeNames = processInstance.findActiveActivityNames();

		System.out.println(activeNames);

	}
	
	/**
	 * 执行drive节点
	 */
	public void test05Exedrive() {

		ProcessInstance processInstance = executionService
				.createProcessInstanceQuery().processInstanceId(
						"forkjoin.10001").uniqueResult();

		String eid = processInstance.findActiveExecutionIn("drive").getId();

		processInstance = executionService.signalExecutionById(eid);

		Set<String> activeNames = processInstance.findActiveActivityNames();

		System.out.println(activeNames.size());
		
		System.out.println(processInstance.isEnded());

	}

}

 

执行test02startFork()发起流程的时候，会有3个state节点同时等待着被处理，test04ExeSendEmail()执行的后，sendEmail的state继续往下执行，因为此时sendMSN的state并未执行，所以join1的join现在处于等待状态。test04ExeSendMSN()执行后sendMSN的state被执行了，所以这个时候join1不在处于等待状态了，而直接往下走到了drive。test04ExeQQ()执行后QQ的state继续往下走，和刚才的情况一样，因为drive还处于中断等待状态，所有流程一直在join2等待中…………。test05Exedrive()执行后，大功告成，所有节点都已经执行完毕，流程走向终结。