fourinone之WareHouse_JAVA_编程开发_程序员俱乐部

好久没写博客了，最近研究fourinone框架一直没什么进展，直到今天，觉得fourinone中的WareHouse模式可以写一下，并行的程序的确很费脑筋：

class="java">

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledThreadPoolExecutor;
/**
 * 这是一个线程执行器
 * @author Administrator
 * 
 */
class Executor //java.io.Closeable
{
	private static ThreadPoolExecutor tpe;
	private static ScheduledThreadPoolExecutor stpe;
	
	//使用可用的几个池线程之一来执行任务
	static ThreadPoolExecutor tpe()
	{
		if(tpe==null)
		{
			int corePoolSize = 10;
			int maximumPoolSize =100;
			long keepAliveTime = 3000;
			TimeUnit unit = TimeUnit.MILLISECONDS;
			BlockingQueue<Runnable> waitQueue = new ArrayBlockingQueue<Runnable>(2000);
			RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();//ThreadPoolExecutor.CallerRunsPolicy();
			/**
			*corePoolSize:核心线程数
			*maximumPoolSize:最大线程数
			*keepAliveTime:当线程数大于corePoolSize时，终止多余的线程所等待的时间
            *unit:keepAliveTime的单位
			*waitQueue:要执行的任务队列
			*handler:线程被阻塞时使用的处理程序
			*/
			tpe =new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, waitQueue, handler);
		}
		return tpe;
	}
	
	static ScheduledThreadPoolExecutor stpe()
	{
		if(stpe==null)
		{
			int corePoolSize = 10;
			stpe =new ScheduledThreadPoolExecutor(corePoolSize);
		}
		return stpe;
	}
	/**
	 * 该方法当t<=0时，执行d任务，当t>0时，执行d任务和i任务，其中i是定时的任务
	 * @param d 非定时执行的任务
	 * @param i 定时执行的任务
	 * @param t 定时开始的时间
	 */
	static void execute(Runnable d, Runnable i, long t){
		tpe().execute(d);
		if(t>0)
			stpe().schedule(i,t,TimeUnit.SECONDS);//在t秒之后执行任务i
	}
	
	static void close(){
		if(tpe!=null){
			try{
				tpe.shutdown();
				tpe=null;
			}catch(SecurityException se){
				//LogUtil.info("[tpe]", "[close]", "[Error Exception:]", se);
			}
		}
		if(stpe!=null){
			try{
				stpe.shutdown();
				stpe=null;
			}catch(SecurityException se){
				//LogUtil.info("[stpe]", "[close]", "[Error Exception:]", se);
			}
		}
	}
}

上面程序中用到的类没啥好说的，是JDK中的并行编程常见的类。

import java.util.*;
import java.util.concurrent.*;
public class WareHouse extends HashMap
{
	public final static int NOTREADY=1,READY=0,EXCEPTION=-1;
	int status=NOTREADY;
	boolean ready = true;
	boolean mark = true; //标记仓库是否被取过
	public WareHouse(){
		super();
	}
	public WareHouse(boolean ready)
	{
		super();
		this.ready=ready;
		status=NOTREADY;
	}
	public int getStatus()
	{
		return status;
    }
	public String getStatusName()
	{
		String[] statusName = new String[]{"EXCEPTION","READY","NOTREADY"};
		return statusName[status+1];
	}
	synchronized void setStatus(int status)
	{
		this.ready = true;	
		this.status = status;
	}
	
	public synchronized boolean isReady()//throws Throwable
	{
		return ready;
	}
	
	public void setMark(boolean m)
	{
		this.mark=m;
	}

	public boolean getMark()
	{
		return mark;
	}
	
}

WareHouse是fourinone中的一个基本的类，这个类是LinkedHashMap的子类，同时又包括几个起到标示作用的私有变量，ready用于标记该仓库是否处理完毕，status代表处理完毕后仓库的状态，mark的作用则在下面的代码中理解。实际上是该仓库的内容是否已经被取过的意思。下面的代码是重点（红色部分）

import java.util.concurrent.*;
import java.util.*;
public class Test extends Executor
{
	public static void main(String [] args) 
	{
//以下为红色
		final WareHouse [] w=new WareHouse[3];    

		for(int j=0;j<3;)
		{
			for(int i=0;i<3;i++)
			{			
				if(w[i]==null)
				{
					System.out.println("启动任务"+i);
					w[i]=doTask();
				}
				else if(w[i].isReady()&&w[i].getMark())
				{
					System.out.println("任务"+i+"完成");
					w[i].setMark(false);
					j++;
				}
			} 
			
		//以上为红色
System.out.println(w[0].get("pi"));
		System.out.println(w[1].get("pi"));
		System.out.println(w[2].get("pi"));
	}
	public static WareHouse doTask()
	{
		final double m=1.0;
		final double n=250000000.0;
		final WareHouse wh=new WareHouse(false);
		execute(new Runnable(){	public void run(){
				 double pi=0.0;
				for(double i=m;i<n;i++)
				{
					pi+=Math.pow(-1,i+1)/(2*i-1);
			
				}
				wh.put("pi",4*pi);
				wh.setStatus(WareHouse.READY);
				}
		},new Runnable(){public void run(){}},0);
		return wh;
	}

}

红色部分的代码值得大家好好的品味一下，在语句final WareHouse [] w=new WareHouse[3];    之后，w[0],w[1],w[2]是等于与null的，这就是为什么有语句
if(w[i]==null)了，一旦if(w[i]==null)，则w[i]=doTask();，而doTask()的第一件事就是让w[i]!=null,这就保证了每个w[i]都仅仅关联一个仓库，而且这个仓库是由线程向里面写入东西的，一旦东西写入完毕，状态改变wh.setStatus(WareHouse.READY);，所以我们在红色区域要一直判断else if(w[i].isReady()&&w[i].getMark())
，为什么能够这样来实现，归根结底在于java一切皆引用在作怪，用于WareHouse的每个实例都是引用，才保证了不论函数如何嵌套调用，本质都是处理同一个内存区域，变化的仅仅是只想这个内存区域的指针地址。
这是我从fourinone中抽象出来的简单代码，供大家参考。