一.    它要能適應不同類型的請求:
本節用 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();
  }
}
看看結果吧.你可以把不同類型的請求不斷地向處理器發送,處理器會不斷地接收請求,放到隊列中,并同時不斷從隊列中提出請求進行處理.