Java 异步消息处理
在前一节实现异步调用的基础上,现在我们来看一下一个完善的Java异步消息处理机制.
[写在本节之前]
在所有这些地方,我始终没有提到设计模式这个词,而事实上,多线程编程几乎每一步都在应该设计模式.你只要能恰如其份地应用它,为什么要在意你用了某某名称的模式呢?
一个说书人它可以把武功招数说得天花乱坠,引得一班听书客掌声如雷,但他只是说书的.真正的武林高手也许并不知道自己的招式在说书人口中叫什么,这不重要,重要的是他能在最恰当的时机把他不知名的招式发挥到极致!
你了解再多的设计模式,或你写出了此类的著作,并不重要,重要的是你能应用它设计出性能卓越的系统.
本节的这个例子,如果你真正的理解了,不要怀疑自己,你已经是Java高手的行列了.如果你抛开本节的内容,五天后能自己独立地把它再实现一次,那你完全可以不用再看我写的文章系列了,至少是目前,我再也没有更高级的内容要介绍了.
上节的Java异步调用为了简化调用关系,很多角色被合并到一个类中实现,为了帮助大家改快地抓住核心的流程.那么一个真正的异步消息处理器,当然不是这样的简单.
一. 它要能适应不同类型的请求:
本节用 makeString来说明要求有返回值的请求.用displayString来说明不需要返回值的请求.
二. 要能同时并发处理多个请求,并能按一定机制调度:
本节将用一个队列来存放请求,所以只能按FIFO机制调度,你可以改用LinkedList,就可以简单实现一个优先级(优先级高的addFirst,低的addLast).
三. 有能力将调用的边界从线程扩展到机器间(RMI)
四. 分离过度耦合,如分离调用句柄(取货凭证)和真实数据的实现.分离调用和执行的过程,可以尽快地将调返回.
现在看具体的实现:
public interface Axman {
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)();
}
简单说c/c++中将要被调用的涵数可以被保存起来,然后去访问,调用,而Java中,我们无法将一个方法保存,除了直接调用,所以将要调用的方法用子类来实现,然后把这些子类实例保存起来,然后在这些子类的实现上调用方法:
interface My{
void test();
}
class A implements My{
public void test(){
System.out.println(“A”):
}
}
class B implements My{
public void test(){
System.out.println(“B”):
}
}
class C implements My{
public void test(){
System.out.println(“C”):
}
}
class MyStruct {
int mark;
My m;
public MyStruct(int mark,My m){this.mark = amrk;this.m = m}
}
数组:
{ new MyStruct(1,new A()),new MyStruct(2,new B()),new MyStruct(3,new C())}
for(xxxxxxxxx) if(参数 ==数组[i].mark) 数组[i].m.test();
这样把要调用的方法转换为对象的保程不仅仅是可以对要调用的方法进行调度,而且可以把对象序列化后在另一台机器上执行,这样就把调用边界从线程扩展到了机器.
回到我们的例子:
class Proxy implements Axman{
private final Scheduler scheduler;
private final Servant servant;
public Proxy(Scheduler scheduler,Servant servant){
this.scheduler = scheduler;
this.servant = servant;
}
public Result resultTest(int count,char c){
FutureResult futrue = new FutureResult();
this.scheduler.invoke(new ResultRequest(servant,futrue,count,c));
return futrue;
}
public void noResultTest(String str){
this.scheduler.invoke(new NoResultRequest(this.servant,str));
}
}
其中scheduler是管理对调用的调度, servant是真正的对方法的执行:
Servant就是去真实地实现方法:
class Servant implements Axman{
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的事了,就象一个人点燃爆竹的引信就跑了,至于那个爆竹什么时候爆炸就不是他能控制的了.
public class Scheduler extends Thread {
private final ActivationQueue queue;
public Scheduler(ActivationQueue 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调用,你要实更复杂的调度就要在这里重新实现:
class ActivationQueue{
private static final int MAX_METHOD_REQUEST = 100;
private final MethodRequest[] requestQueue;
private int tail;
private int head;
private int count;
public ActivationQueue(){
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.head = (this.head + 1) % this.requestQueue.length;
count --;
this.notifyAll();
return request;
}
}
为了将方法调用转化为对象,我们通过实现MethodRequest对象的execute方法来方法具体方法转换成具体对象:
abstract class MethodRequest{
protected final Servant servant;
protected final FutureResult future;
protected MethodRequest(Servant servant,FutureResult future){
this.servant = servant;
this.future = future;
}
public abstract void execute();
}
class ResultRequest extends MethodRequest{
private final int count;
private final char c;
public ResultRequest(Servant servant,FutureResult future,int count,char c){
super(servant,future);
this.count = count;
this.c = c;
}
public void execute(){
Result result = servant.resultTest(this.count,this.c);
this.future.setResult(result);
}
}
class NoResultRequest extends MethodRequest{
private String str;
public NoResultRequest(Servant servant,String str){
super(servant,null);
this.str = str;
}
public void execute(){
this.servant.noResultTest(str);
}
}
而返回的数据我们也将真实数据的获取和取货凭证逻辑分离:
package com.axman.jasync;
public abstract class Result {
public abstract Object getResultValue();
}
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();
}
}
class RealResult extends Result{
private final Object resultValue;
public RealResult(Object resultValue){
this.resultValue = resultValue;
}
public Object getResultValue(){
return this.resultValue;
}
}
OK,现在这个异步消息处理器已经有了模型,这个异步处理器中有昭雪些对象参与呢?
Servant 忠心做真实的事务
ActivationQueue将请求缓存起来以便调度
Scheduler对容器中的请求根据一定原则进行调度执行
Proxy将特定方法请求转换为特定对象
所有这些都是这个异步处理器的核心部件,虽然是核心部件,我们就要进行封装而不能随便让调用者来修改,所以我们用工厂模式(我KAO,我实在不想提模式但有时找不到其它词来表述)来产生处理器Axman对象:
package com.axman.jasync;
public class AxmanFactory {
public static Axman createAxman() {
Servant s = new Servant();
ActivationQueue queue = new ActivationQueue();
Scheduler st = new Scheduler(queue);
Proxy p = new Proxy(st,s);
st.start();
return p;
}
}
好了,我们现在用两个请求的产生者不停产生请求:
ResultInvokeThreadv 发送有返回值的请求:
package com.axman.jasync;
public class ResultInvokeThread extends Thread{
private final Axman ao;
private final char c;
public ResultInvokeThread(String name,Axman ao){
this.ao = ao;
this.c = name.charAt(0);
}
public void run(){
try{
int i = 0;
while(true){
Result result = this.ao.resultTest(i++,c);
Thread.sleep(10);
String value = (String)result.getResultValue();
System.out.println(Thread.currentThread().getName() + " value = " + value);
}
}
catch(Throwable t){}
}
}
NoResultInvokeThread发送无返回值的请求:
package com.axman.jasync;
public class NoResultInvokeThread extends Thread{
private final Axman ao;
public NoResultInvokeThread(String name,Axman ao){
super(name);
this.ao = ao;
}
public void run(){
try{
int i = 0;
while(true){
String s = Thread.currentThread().getName() + i++;
ao.noResultTest(s);
Thread.sleep(20);
}
}
catch(Throwable t){}
}
}
对了,我们还需要一个什么东西来产生一个演示:
package com.axman.jasync;
public class Program {
public static void main(String[] args) {
Axman ao = AxmanFactory.createAxman();
new ResultInvokeThread("Axman",ao).start();
new ResultInvokeThread("Sager",ao).start();
new NoResultInvokeThread("Macke",ao).start();
}
}
看看结果吧.你可以把不同类型的请求不断地向处理器发送,处理器会不断地接收请求,放到队列中,并同时不断从队列中提出请求进行处理.
