多线程编程 高级主题(二)

Java 异步消息处理

在前一节实现异步调用的基础上,现在我们来看一下一个完善的Java异步消息处理机制.

[写在本节之前]

在所有这些地方,我始终没有提到设计模式这个词,而事实上,多线程编程几乎每一步都在应用
设计模式.你只要能恰如其份地应用它,为什么要在意你用了某某名称的模式呢?

一个说书人它可以把武功招数说得天花乱坠,引得一班听书客掌声如雷,但他只是说书的.真正的
武林高手也许并不知道自己的招式在说书人口中叫什么,这不重要,重要的是他能在最恰当的时机把他
不知名的招式发挥到极致!

你了解再多的设计模式,或你写出了此类的著作,并不重要,重要的是你能应用它设计出性能卓越
的系统.

本节的这个例子,如果你真正的理解了,不要怀疑自己,你已经是Java高手的行列了.如果你抛开
本节的内容,五天后能自己独立地把它再实现一次,那你完全可以不用再看我写的文章系列了,至少是
目前,我再也没有更高级的内容要介绍了.

在我的这个异步消息处理器组件中，使用了很多模式，或许我不能准确地命名它们，但我可以
把它们正确地应用到这个组件的实现上，包括这个组件本身，我给它命名为“主动服务模式”，因为组件
自身拥有自己独立的线程来配合客户端请求做出响应，在这个过程中，组件本身有极大的主动性在控制
整个调用过程，而不是依靠客户端的调用者来控制整个过程。

上节的Java异步调用为了简化调用关系,很多角色被合并到一个类中实现,为了帮助大家改快地
抓住核心的流程.那么一个真正的异步消息处理器,当然不是这样的简单.

1. 它要能适应不同类型的请求:
本节用 ResultTest来说明要求有返回值的请求.用noResultTest来说明不需要返回值的请求.

2. 要能同时并发处理多个请求,并能按一定机制调度:
本节将用一个队列来存放请求,所以只能按FIFO机制调度,你可以改用LinkedList,就可以简单实现一个
优先级(优先级高的addFirst,低的addLast).

3. 有能力将调用的边界从线程扩展到机器间(RMI)

4. 分离过度耦合,如分离调用句柄(取货凭证)和真实数据的实现.分离调用和执行的过程,可以尽快地将调返回.

现在看具体的实现:

public interface ISmartAxmanService {
	Result resultTest(int count, char c);

	void noResultTest(String str);
}

这个接口有两个方法要实现,就是有返回值的调用resultTest和不需要返回值的调用noResultTest,
我们把这个接口用一个代理类来实现,目的是将方法调用转化为对象,这样就可以将多个请求(多个方法调)
放到一个容器中缓存起来,然后统一处理,因为Java不支持方法指针,所以把方法调用转换为对象,然后在
这个对象上统一执行它们的方法,不仅可以做到异步处理,而且可以将代表方法调用的请求对象序列化后
通过网络传递到另一个机器上执行(RMI).这也是Java回调机制最有力的实现.

一个简单的例子.
如果 1: 做A
如果 2: 做B
如果 3: 做C
如果有1000个情况,你不至于用1000个case吧?以后再增加呢?

所以如果C/C++程序员,会这样实现: (c和c++定义结构不同)
type define struct MyStruct{
int mark;
(*fn) ();
} MyList;
然后你可以声明这个结构数据:
{1,A,2,B,3,C}
做一个循环:
for(i=0;i<length;i++) {
if(数据组[i].mark == 传入的值){
(数据组[i].*fn)();
break;
}
}

简单说c/c++中将要被调用的函数可以被保存起来,然后去访问,调用,而Java中,我们无法将一个方法保存,
除了直接调用,所以将要调用的方法用子类来实现,然后把这些子类实例保存起来,然后在这些子类的实现上
调用方法:
interface Imy{
void test();
}
class A implements Imy{
public void test(){
System.out.println("A"):
}
}

class B implements Imy{
public void test(){
System.out.println("B"):
}
}

class C implements Imy{
public void test(){
System.out.println("C"):
}
}

class MyStruct {
int mark;
Imy m;
public MyStruct(int mark,Imy m){this.mark = amrk;this.m = m}
}
数组:
{ new MyStruct(1,new A()),new MyStruct(2,new B()),new MyStruct(3,new C())}
for(int i=0;i<数组.length;i++)
if(参数 ==数组[i].mark) {
数组[i].m.test();break;
}
这样把要调用的方法转换为对象的保程不仅仅是可以对要调用的方法进行调度,而且可以把对象序列化后
在另一台机器上执行,这样就把调用边界从线程扩展到了机器.

回到我们的例子:

代理其实就是在接口和实现类之间做一个中间人，一个实现如果直接执行接口的方法，我们无法对这些方法
的调用进行控制和跟踪等行为。代理模式的意义在于用户要访问一个实现时其实是先访问这个对象的代理，由
这个代理来决定如何使用实现类，这样可以起到控制，延迟加载，缓存，隔离被代理的真正实现，也可以对
真正的实现的并发性，安全性和其它附加操作进行控制。

对于用户而言，请求时并不知道他请求的是代理对象，而是直接请求接口的操作，所以代理要和实现
类一样来实现类一样“继承”同一接口，在接口的方法中不去真正的实现而是调用实现类的实现。

class Proxy implements ISmartAxmanService {
	private final Scheduler scheduler; // 这个调度者就是代理用来控制实现类的调用的
	private final SmartAxmanServiceImp serviceImp; // 这个实现类完成接口方法的真正实现

	public Proxy(Scheduler scheduler, SmartAxmanServiceImp serviceImp) {
		this.scheduler = scheduler;
		this.serviceImp = serviceImp;
	}

	public Result resultTest(int count, char c) {

		FutureResult futrue = new FutureResult();
		// 如果弱化代理的调度控制，这里就应该象下面注释掉的代码：
		// doBeforeImpResultTest();
		// serviceImp.ResultTest();
		// doAfterImpResultTest();
		// 这里进行了更复杂的控制，所以把实现和调度传递给ResultRequest来执行。
		this.scheduler.invoke(new ResultRequest(serviceImp, futrue, count, c));
		return futrue;
	}

	public void noResultTest(String str) {

		// 同上
		this.scheduler.invoke(new NoResultRequest(this.serviceImp, str));
	}
}

SmartAxmanServiceImp就是去真实地实现方法:

class SmartAxmanServiceImp implements ISmartAxmanService {
	public Result resultTest(int count, char c) {
		char[] buf = new char[count];
		for (int i = 0; i < count; i++) {
			buf[i] = c;
			try {
				Thread.sleep(100); // 模拟耗时操作
			} catch (Throwable t) {}
		}
		return new RealResult(new String(buf));
	}

	public void noResultTest(String str) {
		try {
			System.out.println("displayString :" + str);
			Thread.sleep(10);
		} catch (Throwable t) {}
	}
}

在scheduler将方法的调用(invkoe)和执行(execute)进行了分离,调用就是开始“注册”方法到要执行的容器中,这样
就可以立即返回出来.真正执行多久就是execute的事了,就象一个人点燃爆竹的引信就跑了,至于那个爆竹什么时候
爆炸就不是他能控制的了.
因为这个组件有自己独立的线程在执行调用者的请求，是整个组件“主动性”的关键。事实上，它还实现了Worker Thread

模式和生产者-消费者模式。

class Scheduler extends Thread {

	private final SmartQueue queue; // 所有加载了请求方法的对象需要放在一个容器上以便统一调用处理。
									// 这里实现了一个队列
	public Scheduler(SmartQueue queue) {
		this.queue = queue;
	}

	public void invoke(MethodRequest request) {
		this.queue.putRequest(request);
	}

	public void run() {
		while (true) {
			// 如果队列中有请求线程,则开始执行请求
			MethodRequest request = this.queue.takeRequest();
			request.execute();
		}
	}
}

在scheduler中只用一个队列来保存代表方法和请求对象,实行简单的FIFO调度,你要实更复杂的调度就要在这里重新实现,这个队列其实是生产者与消费者模式中的channel构件:

class SmartQueue {

	private static final int MAX_METHOD_REQUEST = 100;
	private final MethodRequest[] requestQueue;
	private int tail;
	private int head;
	private int count;

	public SmartQueue() {
		this.requestQueue = new MethodRequest[MAX_METHOD_REQUEST];
		this.head = this.count = this.tail = 0;
	}

	public synchronized void putRequest(MethodRequest request) {
		while (this.count >= this.requestQueue.length) {
			try {
				this.wait();
			} catch (Throwable t) {}
		}
		this.requestQueue[this.tail] = request;
		tail = (tail + 1) % this.requestQueue.length; //如果到数组最后则从头开始
		count++;
		this.notifyAll();
	}

	public synchronized MethodRequest takeRequest() {
		while (this.count <= 0) {
			try {
				this.wait();
			} catch (Throwable t) {
			}
		}
		MethodRequest request = this.requestQueue[this.head];

		// this.requestQueue[this.head] = null;
		// 考虑注释掉的这一行代码，对于循环的队列还不是非常明显，如果是非循环的队列。这个元素
		// 可能永远被持有，比如一个Stack中如果put了，再pop时，虽然对象被取走了，栈顶指针指象了index+1;
		// 但底层的数组中那个index位置的对象还被数组本身引用着，如果没有put动作替换这个位置的句柄指向一个
		// 新的对象，则已经pop出去的对象一直被数组本身引用着。所以对于处于数据结构中的对象如果要从中取出，
		// 数据结果本身要取消对它的引用。

		this.head = (this.head + 1) % this.requestQueue.length;
		count--;
		this.notifyAll();
		return request;
	}
}

为了将方法调用转化为对象,我们通过实现MethodRequest对象的execute方法来方法具体方法转换成具体对象:

public abstract class MethodRequest {

	protected final SmartAxmanServiceImp serviceImp;
	protected final FutureResult future;

	protected MethodRequest(SmartAxmanServiceImp serviceImp, FutureResult future) {
		this.serviceImp = serviceImp;
		this.future = future;
	}

	public abstract void execute();

}

public class ResultRequest extends MethodRequest {
	private final int count;
	private final char c;

	public ResultRequest(SmartAxmanServiceImp serviceImp, 
			FutureResult future,
			int count, 
			char c) {
		super(serviceImp, future);
		this.count = count;
		this.c = c;
	}

	public void execute() {
		Result result = serviceImp.resultTest(this.count, this.c);
		this.future.setResult(result);
	}
}

public class NoResultRequest extends MethodRequest {
	private String str;

	public NoResultRequest(SmartAxmanServiceImp serviceImp, String str) {
		super(serviceImp, null);
		this.str = str;
	}

	public void execute() {
		this.serviceImp.noResultTest(str);
	}
}

而返回的数据我们也将真实数据的获取和取货凭证逻辑分离，这里虽然是经典的Future模式，但我们可以看到，
FutureResult其本质也是Result与RealResult的代理，或桥梁。

public abstract class Result {
	public abstract Object getResultValue();
}

public class FutureResult extends Result {

	private Result result;
	private boolean completed;

	public synchronized void setResult(Result result) {
		this.result = result;
		this.completed = true;
		this.notifyAll();
	}

	public synchronized Object getResultValue() {
		while (!this.completed) {
			try {
				this.wait();
			} catch (Throwable t) {}
		}
		return this.result.getResultValue();
	}
}

public class RealResult extends Result {

	private final Object resultValue;

	public RealResult(Object resultValue) {
		this.resultValue = resultValue;
	}

	public Object getResultValue() {
		return this.resultValue;
	}
}

OK,现在这个异步消息处理器已经有了模型,这个异步处理器中有哪些对象参与呢?

SmartAxmanServiceImp 忠心做真实的事务

SmartQueue将请求缓存起来以便调度

Scheduler对容器中的请求根据一定原则进行调度执行

Proxy将特定方法请求转换为特定对象

所有这些都是这个异步处理器的核心部件,既然是核心部件,我们就要进行封装而不能随便让调用者来修改,所以我们
用工厂模式(我KAO,我实在不想提模式但有时找不到其它词来表述)来产生处理器ISmartAxmanService对象:

class SmartAxmanServiceFactory {
	public static synchronized ISmartAxmanService createService() {
		SmartAxmanServiceImp imp = new SmartAxmanServiceImp();
		SmartQueue queue = new SmartQueue();
		Scheduler st = new Scheduler(queue);
		Proxy p = new Proxy(st, imp);
		st.start();
		return p;
	}

}

好了,我们现在用两个请求的产生者不停产生请求:

ResultInvokeThread 发送有返回值的请求:

public class ResultInvokeThread extends Thread {
	private final ISmartAxmanService service;
	private final char c;

	public ResultInvokeThread(String name, ISmartAxmanService service) {
		this.service = service;
		this.c = name.charAt(0);
	}

	public void run() {
		try {
			int i = 0;
			while (true) {
				Result result = this.service.resultTest(i++, c);
				Thread.sleep(10);
				String value = (String) result.getResultValue();
				System.out.println(Thread.currentThread().getName()
						+ " value = " + value);
			}
		} catch (Throwable t) {}
	}

}

NoResultInvokeThread发送无返回值的请求:

public class NoResultInvokeThread extends Thread {

	private final ISmartAxmanService service;

	public NoResultInvokeThread(String name, ISmartAxmanService service) {
		super(name);
		this.service = service;
	}

	public void run() {
		try {
			int i = 0;
			while (true) {
				String s = Thread.currentThread().getName() + i++;
				service.noResultTest(s);
				Thread.sleep(20);
			}
		}
		catch (Throwable t) {
		}
	}
}

对了,我们还需要一个什么东西来产生一个演示:

public class MainTest {
	public static void main(String[] args) throws Exception {
		ISmartAxmanService service = SmartAxmanServiceFactory.createService();
		new ResultInvokeThread("Axman",service).start();
	    new ResultInvokeThread("Sager",service).start();
	    new NoResultInvokeThread("Macke",service).start();
	}
}

看看结果吧.你可以把不同类型的请求不断地向处理器发送,处理器会不断地接收请求,放到队列中,并同时不断从队列
中提出请求进行处理.

用了四种工具才将本文的类图，序列图和时绪图画好。