Java 异步消息处理.doc

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 MyStructint mark;(*fn) (); MyList; 然后你可以声明这个结构数据: 1,A, 2,B 3,C做一个循环:for(i=0;ilength;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 charcount; for(int i = 0;i = this.requestQueue.length) try this.wait(); catch (Throwable t) this.requestQueuethis.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.requestQueuethis.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 getResult();class FutureResult extends Result private Result result; private boolean completed; public synchronized void setResult(Result result) this.result = result; pleted = true; this.notifyAll(); public synchronized Object getResult() while(!pleted) try this.wait(); catch(Throwable t) return this.result.getResult(); class RealResult extends Result private final Object result; public RealResult(Object result) this.result = result; public Object getResult() return this.result; 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 = (String)result.getResult(); System.out.println(Thread.currentThread().getName() + = + ); 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(); 看看结果吧.你可以把不同类型的请求不断地向处理器发送,处理器会不断地接收请求,放到队列中,并同时不断从队列中提出请求进行处理. 在调用一个方法的时候，程序会进入被调用方法体内，执行完这个被调用方法后，才返回执行下一条语句。怎么做到像ajax异步请求一样，发送请求后，没等请求响应就执行下一条语句呢?对于java的异步请求，找了许多教材都没有找到，如thinking in java、core java2 .等等。受多线程下载工具以及mootools的Request的启发，做了一个java版的Request，不知它的性能如何。 Request:请求载体public class Request private RequestContent rc;/请求主体 public