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 getResult();
}

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 getResult(){
    while(!this.completed){
      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();
  }
}

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

评论

1 楼 snail_gtt 2009-07-08  

这个例子很好，目前我遇到的一个问题正好和这个方案相当吻合。受教了~

不过有一个问题，在这里我先提出来。

首先我发了很多请求过来了，然后由ActivationQueue缓存起来了。
如果在这些请求尚未处理完成时，服务器停了，再启动后，这些未完成的请求就没有了。

这种情况是否意味着，需要将请求持久化？