java异步处理

piperzero

浏览: 3553917 次
性别:
来自: 杭州

最近访客更多访客>>

handsome1234

u012363178

zhangketuan

wangyy

博主相关

博客

微博

相册

留言

关于我

文章分类

全部博客 (3796)

社区版块

存档分类

一.它要能适应不同类型的请求:
本节用makeString来说明要求有返回值的请求.用displayString来说明不需要返回值的请求.
二.要能同时并发处理多个请求,并能按一定机制调度:
本节将用一个队列来存放请求,所以只能按FIFO机制调度,你可以改用LinkedList,就可以简单实现一个优先级(优先级高的addFirst,低的addLast).
三.有能力将调用的边界从线程扩展到机器间(RMI)
四.分离过度耦合,如分离调用句柄(取货凭证)和真实数据的实现.分离调用和执行的过程,可以尽快地将调返回.

现在看具体的实现:
publicinterfaceAxman{
ResultresultTest(intcount,charc);
voidnoResultTest(Stringstr);
}
这个接口有两个方法要实现,就是有返回值的调用resultTest和不需要返回值的调用
noResultTest, 我们把这个接口用一个代理类来实现,目的是将方法调用转化为对象,这样就可以将多个请求(多个方法调)放到一个容器中缓存起来,然后统一处理,因为 Java不支持方法指针,所以把方法调用转换为对象,然后在这个对象上统一执行它们的方法,不仅可以做到异步处理,而且可以将代表方法调用的请求对象序列化后通过网络传递到另一个机器上执行(RMI).这也是Java回调机制最有力的实现.
一个简单的例子.
如果1:做A
如果2:做B
如果3:做C
如果有1000个情况,你不至于用1000个case吧?以后再增加呢?
所以如果C/C++程序员,会这样实现:(c和c++定义结构不同)

typedefinestructMyStruct{
intmark;
(*fn)();
}MyList;

然后你可以声明这个结构数据:
{1,A,
2,B
3,C
}
做一个循环:
for(i=0;i<length;i++){
if(数据组[i].mark==传入的值)(数据组[i].*fn)();
}
简单说c/c++中将要被调用的涵数可以被保存起来,然后去访问,调用,而Java中,我们无法将一个方法保存,除了直接调用,所以将要调用的方法用子类来实现,然后把这些子类实例保存起来,然后在这些子类的实现上调用方法:
interfaceMy{
voidtest();
}

classAimplementsMy{
publicvoidtest(){
System.out.println(“A”):
}
}
classBimplementsMy{
publicvoidtest(){
System.out.println(“B”):
}
}

classCimplementsMy{
publicvoidtest(){
System.out.println(“C”):
}
}

classMyStruct{

intmark;
Mym;
publicMyStruct(intmark,Mym){this.mark=amrk;this.m=m}
}
数组:
{newMyStruct(1,newA()),newMyStruct(2,newB()),newMyStruct(3,newC())}
for(xxxxxxxxx)if(参数==数组[i].mark)数组[i].m.test();

这样把要调用的方法转换为对象的保程不仅仅是可以对要调用的方法进行调度,而且可以把对象序列化后在另一台机器上执行,这样就把调用边界从线程扩展到了机器.

回到我们的例子:
classProxyimplementsAxman{
privatefinalSchedulerscheduler;
privatefinalServantservant;

publicProxy(Schedulerscheduler,Servantservant){
this.scheduler=scheduler;
this.servant=servant;
}
publicResultresultTest(intcount,charc){
FutureResultfutrue=newFutureResult();
this.scheduler.invoke(newResultRequest(servant,futrue,count,c));
returnfutrue;
}

publicvoidnoResultTest(Stringstr){
this.scheduler.invoke(newNoResultRequest(this.servant,str));
}
}

其中scheduler是管理对调用的调度,servant是真正的对方法的执行:

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

classServantimplementsAxman{
publicResultresultTest(intcount,charc){
char[]buf=newchar[count];
for(inti=0;i<count;i++){
buf[i]=c;
try{
Thread.sleep(100);
}catch(Throwablet){}
}
returnnewRealResult(newString(buf));
}

publicvoidnoResultTest(Stringstr){
try{
System.out.println("displayString:"+str);
Thread.sleep(10);
}catch(Throwablet){}
}
}
在scheduler 将方法的调用(invkoe)和执行(execute)进行了分离,调用就是开始”注册”方法到要执行的容器中,这样就可以立即返回出来.真正执行多久就是execute的事了,就象一个人点燃爆竹的引信就跑了,至于那个爆竹什么时候爆炸就不是他能控制的了.
publicclassSchedulerextendsThread{
privatefinalActivationQueuequeue;
publicScheduler(ActivationQueuequeue){
this.queue=queue;
}

publicvoidinvoke(MethodRequestrequest){
this.queue.putRequest(request);
}

publicvoidrun(){
while(true){

//如果队列中有请求线程,测开始执行请求
MethodRequestrequest=this.queue.takeRequest();
request.execute();
}
}
}
在scheduler中只用一个队列来保存代表方法和请求对象,实行简单的FIFO调用,你要实更复杂的调度就要在这里重新实现:
classActivationQueue{
privatestaticfinalintMAX_METHOD_REQUEST=100;
privatefinalMethodRequest[]requestQueue;
privateinttail;
privateinthead;
privateintcount;

publicActivationQueue(){
this.requestQueue=newMethodRequest[MAX_METHOD_REQUEST];
this.head=this.count=this.tail=0;
}

publicsynchronizedvoidputRequest(MethodRequestrequest){
while(this.count>=this.requestQueue.length){
try{
this.wait();
}
catch(Throwablet){}
}
this.requestQueue[this.tail]=request;
tail=(tail+1)%this.requestQueue.length;
count++;
this.notifyAll();

}

publicsynchronizedMethodRequesttakeRequest(){
while(this.count<=0){
try{
this.wait();
}
catch(Throwablet){}

}

MethodRequestrequest=this.requestQueue[this.head];
this.head=(this.head+1)%this.requestQueue.length;
count--;
this.notifyAll();
returnrequest;
}
}

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

protectedMethodRequest(Servantservant,FutureResultfuture){
this.servant=servant;
this.future=future;
}

publicabstractvoidexecute();
}

classResultRequestextendsMethodRequest{
privatefinalintcount;
privatefinalcharc;
publicResultRequest(Servantservant,FutureResultfuture,intcount,charc){
super(servant,future);
this.count=count;
this.c=c;
}
publicvoidexecute(){
Resultresult=servant.resultTest(this.count,this.c);
this.future.setResult(result);
}
}

classNoResultRequestextendsMethodRequest{
privateStringstr;
publicNoResultRequest(Servantservant,Stringstr){
super(servant,null);
this.str=str;
}

publicvoidexecute(){
this.servant.noResultTest(str);
}
}

而返回的数据我们也将真实数据的获取和取货凭证逻辑分离:
packagecom.axman.jasync;

publicabstractclassResult{
publicabstractObjectgetResult();
}

classFutureResultextendsResult{
privateResultresult;
privatebooleancompleted;

publicsynchronizedvoidsetResult(Resultresult){
this.result=result;
this.completed=true;
this.notifyAll();
}

publicsynchronizedObjectgetResult(){
while(!this.completed){
try{
this.wait();
}catch(Throwablet){}
}
returnthis.result.getResult();
}
}

classRealResultextendsResult{
privatefinalObjectresult;

publicRealResult(Objectresult){
this.result=result;
}
publicObjectgetResult(){
returnthis.result;
}
}
OK,现在这个异步消息处理器已经有了模型,这个异步处理器中有昭雪些对象参与呢?
Servant忠心做真实的事务
ActivationQueue将请求缓存起来以便调度
Scheduler对容器中的请求根据一定原则进行调度执行
Proxy将特定方法请求转换为特定对象
所有这些都是这个异步处理器的核心部件,虽然是核心部件,我们就要进行封装而不能随便让调用者来修改,所以我们用工厂模式(我KAO,我实在不想提模式但有时找不到其它词来表述)来产生处理器Axman对象:
packagecom.axman.jasync;

publicclassAxmanFactory{
publicstaticAxmancreateAxman(){
Servants=newServant();
ActivationQueuequeue=newActivationQueue();
Schedulerst=newScheduler(queue);
Proxyp=newProxy(st,s);
st.start();
returnp;
}
}
好了,我们现在用两个请求的产生者不停产生请求:
ResultInvokeThreadv发送有返回值的请求:
packagecom.axman.jasync;

publicclassResultInvokeThreadextendsThread{
privatefinalAxmanao;
privatefinalcharc;
publicResultInvokeThread(Stringname,Axmanao){
this.ao=ao;
this.c=name.charAt(0);
}

publicvoidrun(){
try{
inti=0;
while(true){
Resultresult=this.ao.resultTest(i++,c);
Thread.sleep(10);
String=(String)result.getResult();
System.out.println(Thread.currentThread().getName()+"="+);
}
}
catch(Throwablet){}
}
}

NoResultInvokeThread发送无返回值的请求:
packagecom.axman.jasync;

publicclassNoResultInvokeThreadextendsThread{
privatefinalAxmanao;
publicNoResultInvokeThread(Stringname,Axmanao){
super(name);
this.ao=ao;
}

publicvoidrun(){
try{
inti=0;
while(true){
Strings=Thread.currentThread().getName()+i++;
ao.noResultTest(s);
Thread.sleep(20);
}
}
catch(Throwablet){}
}
}

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

publicclassProgram{
publicstaticvoidmain(String[]args){
Axmanao=AxmanFactory.createAxman();
newResultInvokeThread("Axman",ao).start();
newResultInvokeThread("Sager",ao).start();
newNoResultInvokeThread("Macke",ao).start();
}
}
看看结果吧.你可以把不同类型的请求不断地向处理器发送,处理器会不断地接收请求,放到队列中,并同时不断从队列中提出请求进行处理.

在调用一个方法的时候，程序会进入被调用方法体内，执行完这个被调用方法后，才返回执行下一条语句。怎么做到像ajax异步请求一样，发送请求后，没等请求响应就执行下一条语句呢?对于java的异步请求，找了许多教材都没有找到，如thinking in java、core java2 ......等等。受多线程下载工具以及mootools的Request的启发，做了一个java版的Request，不知它的性能如何。

Request:请求载体

public class Request {
private RequestContent rc;//请求主体
public Request(RequestContent rc){
this.rc=rc;
}
protected void start(){ //开始请求
final Thread t=new Thread(new Runnable(){
public void run(){
try{
rc.doSomeThing();//响应请求
}catch (Exception e) {
e.printStackTrace();
rc.onFailure();//如果执行失败
}
rc.onSuccess();//如果执行成功
}}
);
t.start();
}
}

RequestContent:请求主体

abstract class RequestContent {
void onSuccess(){//执行成功的动作。用户可以覆盖此方法
System.out.println("onSuccess");
}
void onFailure(){ //执行失败的动作。用户可以覆盖此方法
System.out.println("onFailure");
}
abstract void doSomeThing(); //用户必须实现这个抽象方法，告诉子线程要做什么
}

Test:测试

new Request(new RequestContent(){
void doSomeThing(){
System.out.println("doSomething");
}
void onSuccess(){
System.out.println("override onSuccess");
}
}).start();

分享到：