Active Object

2012到2014

浏览: 68766 次

最近访客更多访客>>

sanshaoye7_510

huazhong3211

sxgn0423

低头麦穗

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

java多线程设计模式

Active Object Pattern是由很多组件构成的，是一个大型的Pattern。适合大规模问题。

首先由Client发出请求，Active Object的参与者配合得到请求的返回值。

Active Object中，首先异步消息都传递给Proxy，Proxy将这些消息放在队列里面，然后由一个线程（Scheduler）不断的获取这些消息，让这些消息交给“提货单”，真正的消息处理工作交给Servant，Servant把结果交给“提货单”，最后结果返回到Client。

下面的示例展示了这个过程（不过是一个小规模的问题）。

示例中：消息有makeString和displayString，首先是一个生产者和消费者模型，队列中没消息时获取消息就得等待，队列满了时插入消息也要等待。displayString的工作采用了Thread Per Message Pattern，makeString的工作则用FuturePattern。

下面是详细代码：

Result类是一个虚拟数据参与者，它有两个子类：Future参与者FutureResult和RealResult。这里实际用到了Future Pattern。（FutureResult中一旦设置了Result，马上通知别人（提货单）可以获取Result，“提货单”是MakeStringRequest，注意其中的：future.setResult(result);）。

package activeobject;
public abstract class Result {
	public abstract Object getResultValue();
}

package activeobject;
public class RealResult extends Result {
	private final Object resultValue;
	public RealResult(Object resultValue){
		this.resultValue=resultValue;
	}
	@Override
    public Object getResultValue(){
    	return resultValue;
    }
}

package activeobject;
public class FutureResult extends Result {
	private Result result;
	public boolean ready=false;
	public synchronized void setResult(Result result){
		this.result=result;
		this.ready=true;
		notifyAll();
	}
	public synchronized Object getResultValue(){
		while(!ready){
			try{
				wait();
			}catch(InterruptedException e){
				e.printStackTrace();
			}
		}
		return result.getResultValue();
	}
}

        MethodRequest类是一个抽象化的请求。具体的请求是其子类MakeStringRequest、DisplayRequest。
MethodRequest中的字段servant用来进行实际处理的Servant实例，字段future用来存放设置返回值的FutureResult实例，如果请求不需要返回，这个域为空。servant和future都可以让子类来处理，故声明为protected。
        MakeStringRequest会设置servant和future，执行execute时候让servant生成一个Result，这个Result是RealResult，是makeString的结果，makeString是一个需要时间的操作，操作完才能得到RealResult。得到结果后马上设置FutureResult的结果，这个FutureResult之前就是Proxy中定义的，然后返回给Client，但由于没有设置RealResult，它只能陷入等待，所以，得到RealResult后，Client才能继续运行。
        DisplayRequest让servant显示字符串，其实，也就是Client发请求，通过Proxy（将消息传递给队列），队列（请求缓存站）后交给servant执行，也就是Thread Per Message模型了。

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

package activeobject;
public class MakeStringRequest extends MethodRequest{
    private final int count;
    private final char fillchar;
    public MakeStringRequest(Servant servant,FutureResult future,int count ,char fillchar){
    	super(servant,future);
    	this.count=count;
    	this.fillchar=fillchar;
    }
    public void execute(){
    	Result result=servant.makeString(count,fillchar);
    	future.setResult(result);
    }
}

package activeobject;
public class DisplayStringRequest extends MethodRequest {
    private final String string;
    public DisplayStringRequest(Servant servant,String string){
    	super(servant,null);
    	this.string=string;
    }
    public void execute(){
    	servant.displayString(string);
    }
}

接口ActiveObject有产生Result（makeString）和显示字符串的虚方法。

Servant类实现了这个接口，它先生成一个String，然后将String包装成RealResult。（这是一个耗时的过程），以及显示一个字符串。当然，Servant是实际处理请求的类。
Proxy也实现了ActiveObject接口，有字段scheduler和servant。它的makeString是向队列中添加制造请求，它的displayString是向队列中添加显示请求。这个类本质上是将Client的工作与执行的工作分开，充当了“启动”的作用。即“分离方法的启动（invocation）和执行（execution）”

Proxy将Client的请求给SchedulerThread（字段schedule），Scheduler有一个ActivationQueue属性，它有两个工作：将请求添加到队列（ActivationQueue）中；不断的从队列中得到请求，然后执行。

ActivationQueue有一个MethodRequest队列，当队列满了时，添加元素等待，队列为空时，移走一个元素等待。

package activeobject;
public interface ActiveObject {
	public abstract Result makeString(int count,char fillchar);
	public abstract void displayString(String string);
}

package activeobject;
public class Servant implements ActiveObject{
	@Override
	public void displayString(String string) {
		try{
			System.out.println("displayString:"+string);
			Thread.sleep(10);
		}catch(InterruptedException e){
			e.printStackTrace();
		}
	}

	@Override
	public Result makeString(int count, char fillchar) {
		char[] buffer=new char[count];
		for(int i=0;i<count;i++)
			buffer[i]=fillchar;
		try{
			Thread.sleep(100);
		}catch(InterruptedException e){
			e.printStackTrace();
		}
		return new RealResult(new String(buffer));
	}
}

package activeobject;
public class Proxy implements ActiveObject{
	private final SchedulerThread scheduler;
	private final Servant servant;
	public Proxy(SchedulerThread scheduler,Servant servant){
		this.scheduler=scheduler;
		this.servant=servant;
	}
	@Override
	public void displayString(String string) {
		scheduler.invoke(new DisplayStringRequest(servant,string));
	}
	@Override
	public Result makeString(int count, char fillchar) {
		FutureResult future=new FutureResult();
		scheduler.invoke(new MakeStringRequest(servant,future,count,fillchar));
		return future;
	}
}

package activeobject;
public class SchedulerThread extends Thread{
    private final ActivationQueue queue;
    public SchedulerThread(ActivationQueue queue){
    	this.queue=queue;
    }
    public void invoke(MethodRequest request){
    	queue.putRequest(request);
    }
    @Override
    public void run(){
    	while(true){
    		MethodRequest request=queue.takeRequest();
    		request.execute();
    	}
    }
}

package activeobject;
public 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=0;
		this.tail=0;
		this.count=0;
	}
	public synchronized void putRequest(MethodRequest request){
		while(count>=requestQueue.length){
			try{
				wait();
			}catch(InterruptedException e){
				e.printStackTrace();
			}
		}
		requestQueue[tail]=request;
		tail=(tail+1)%requestQueue.length;
	    count++;
	    notifyAll();
	}
	public synchronized MethodRequest takeRequest(){
		while(count<=0){
			try{
				wait();
			}catch(InterruptedException e){
				e.printStackTrace();
			}
		}
		MethodRequest request=requestQueue[head];
		head=(head+1)%requestQueue.length;
	    count--;
	    notifyAll();
	    return request;
	}
}

这里有一个ActiveObjectFactory类，它是为接口提供一个简便的“外观”，它用来产生一个Proxy对象，当然与这个对象相关的Servant、SchedulerThread、Queue都产生了，并且线程SchedulerThread也开启了，这个线程是不断的处理请求，而Client中的线程是不断的发出请求，让Proxy将请求加到队列中。

package activeobject;
public class ActiveObjectFactory {
	public static ActiveObject createActiveObject(){
		Servant servant=new Servant();
		ActivationQueue queue=new ActivationQueue();
		SchedulerThread scheduler=new SchedulerThread(queue);
		Proxy proxy=new Proxy(scheduler,servant);
		scheduler.start();
		return proxy;
	}
}

这个模型中的整个类图如下：

测试类（发出请求的Client）如下：

package activemain;
import activeobject.ActiveObject;
import activeobject.Result;

public class MakerClientThread extends Thread{
	private final ActiveObject activeObject;
	private final char fillchar;
	public MakerClientThread(String name,ActiveObject activeObject){
		super(name);
		this.activeObject=activeObject;
		this.fillchar=name.charAt(0);
	}
	@Override
    public void run(){
		try{
			for(int i=0;true;i++){
				Result result=activeObject.makeString(i, fillchar);
				Thread.sleep(10);
				String value=(String)result.getResultValue();
				System.out.println(Thread.currentThread().getName()+":value="+value);
			}
		}catch(InterruptedException e){
			e.printStackTrace();
		}
	}
}

package activemain;
import activeobject.ActiveObject;

public class DisplayClientThread extends Thread{
	private final ActiveObject activeObject;;
	public DisplayClientThread(String name,ActiveObject activeObject){
		super(name);
		this.activeObject=activeObject;
	}
	@Override
    public void run(){
		try{
			for(int i=0;true;i++){
				String string=Thread.currentThread().getName()+"  "+i;
				activeObject.displayString(string);
				Thread.sleep(200);
			}
		}catch(InterruptedException e){
			e.printStackTrace();
		}
	}

}

package activemain;
import activeobject.ActiveObject;
import activeobject.ActiveObjectFactory;

public class Main {
	public static void main(String[] args){
		ActiveObject activeObject=ActiveObjectFactory.createActiveObject();
		new MakerClientThread("Alice",activeObject).start();
		new MakerClientThread("Bobby",activeObject).start();
		new DisplayClientThread("Chris",activeObject).start();
	}
}

程序执行的结果：

Bobby:value=
displayString:Chris 0
Alice:value=
displayString:Chris 1
Bobby:value=B
Alice:value=A
displayString:Chris 2
Bobby:value=BB
Alice:value=AA
displayString:Chris 3
Bobby:value=BBB
Alice:value=AAA
displayString:Chris 4
Bobby:value=BBBB
Alice:value=AAAA
displayString:Chris 5
Bobby:value=BBBBB
Alice:value=AAAAA
Bobby:value=BBBBBB
displayString:Chris 6
Alice:value=AAAAAA
displayString:Chris 7
Bobby:value=BBBBBBB
Alice:value=AAAAAAA
displayString:Chris 8
Bobby:value=BBBBBBBB
Alice:value=AAAAAAAA
displayString:Chris 9
Bobby:value=BBBBBBBBB
Alice:value=AAAAAAAAA
displayString:Chris 10
Bobby:value=BBBBBBBBBB
displayString:Chris 11
Alice:value=AAAAAAAAAA
Bobby:value=BBBBBBBBBBB
Alice:value=AAAAAAAAAAA
displayString:Chris 12
Bobby:value=BBBBBBBBBBBB

这个模式处理的问题是：当多个Client使用不安全的Servant，且Servant相应的时间比较久，这回拖垮Client的响应性。

查看图片附件

分享到：

多线程设计模式 | Thread Specific Storage

2013-05-05 12:11
浏览 939
评论(0)
分类:编程语言
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

Active Object

评论

发表评论

相关推荐

最近访客 更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

Active Object

评论

发表评论

相关推荐

多线程设计模式

Thread Specific Storage

Two Phase Termination

Future Pattern

Worker Thread

Thread per Message

Read Write Lock

Producer Consumer Pattern

Balking Pattern

Guarded Suspension Pattern

Immutable

Single Threaded Execution

最近访客更多访客>>