Java多线程之工作流网图

工作流网图基于前驱图实现。较之前几个问题，这个问题也是生产中的常见的模型，有人把这个模型称为AOE和AOV。

操作系统原理之进程和线程管理
 Java的多线程和并发知识纲要
 Java多线程之生产者和消费者模式
 Java多线程之哲学家就餐问题
 Java多线程之读者写者问题
 Java多线程之吸烟者问题
 Java多线程之熟睡的理发师问题

AOV和AOE

AOV网与工作流网（AOE）在模型结构上是相似的，它们都是基于图的结构，以节点表示活动，有向边表示流程的流向。

AOV网的有向边仅仅只表示活动的前后次序，也可以说是流程中的流程流向。工作流网中的有向边却不仅如此，它可以在每条边上设置不同的条件来决定活动的下一环节是什么，它的出度就不一定是所有有向边了。

因此，AOV网其实是工作流网（AOE）的一种特例，是一种全入全出的有向无环工作流网。

AOE网的定义是在带权有向图中若以顶点表示事件，有向边表示活动，边上的权值表示该活动持续的时间。
从定义上来看，很容易看出两种网的不同，AOV网的活动以顶点表示，而AOE网的活动以有向边表示。

纵观这两种网图，其实它们总体网络结构是一样的，仅仅是活动所表示的方式不同，因此从AOV网转换成AOE网应该是可行的。

问题描述

在生产中，比如玩具组装车间情景里，有这样的流水线（可能不准确，只是为了说明工作流网）：

工作流网图

某一个节点，比如P3，它需要等待P0和P2的工作完成后才能进行工作。

问题分析

AOE是边上有权值，节点值为“0”（只是为了理解，更准确的是“没有”）。
AOV可以转换成AOE，只要将路径上的权值设置为“0”，节点值设置为非“0”。

在这个问题下，工作流图的图存储：
1、如果使用邻接矩阵的话，存储边上的权值比较方便，但是去寻找节点的出度需要遍历操作。
2、如果使用邻接表的话，存储边上的权值比较麻烦，但是去寻找节点的出度比较方便。

进程之间的工作完成的通知可以使用信号量。

我的解决

我采用的工作流图的图存储方式是邻接矩阵，矩阵内的元素包含权值和信号量，这样既能方便找到某节点的出度和入度到某节点的源节点，也能迅速定位到这条边所使用的信号量。使用邻接表的话则不大容易实现。

将上图的工作流网图加上具体的边上权值，节点值为“0”，如下图：

工作流网图

源码实现

具体的详细源码可以参看文末的链接。

private class NodeThread extends Thread {

	private int id;
	private String name;

	// 该节点生产时间
	private int time;

	// 生产的次数
	private int amount;

	private volatile boolean status;

	public NodeThread(int id, int time) {
		this.id = id;
		this.name = "node-" + id;
		this.time = time;
	}

	@Override
	public void run() {
		try {
			status = true;
			barrier.await();
			System.out.println(name + " is ready");
			long begin = System.currentTimeMillis();
			while (status) {
				System.out.println(name + " wait for semaphores");
				long start = System.currentTimeMillis();
				recv();
				System.out.println(name + " is working");
				process();
				long end = System.currentTimeMillis();
				System.out.println(name + " finish work, cost " + (end - start) + ", amount " + amount);
				send();
				System.out.println(name + " has sent semaphores");
			}
			long finish = System.currentTimeMillis();
			System.out.println(name + " IS DONE, TOTAL COST " + (finish - begin) + ", FINAL AMOUNT " + amount);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	// 等待所有入该节点的信号量
	public void recv() throws InterruptedException {
		for (int i = 0; i < nums; i++) {
			Edge e = diagram[i][id];
			if (e == null)
				continue;
			e.semaphore.P();
		}
	}

	// 通知所有出该节点的信号量
	public void send() {
		for (int i = 0; i < nums; i++) {
			Edge e = diagram[id][i];
			if (e == null)
				continue;
			executor.schedule(new Msg(e.semaphore), e.value, TimeUnit.SECONDS);
		}
	}

	// 生产该节点负责的部分
	public void process() throws InterruptedException {
		Thread.sleep(time * 1000);
		if (++amount == goods)
			status = false;
	}

}