一个任务队列的BlockingQueue实现

一、Concurrent简单介绍

Concurrent是jdk1.5推出来的对多线程实现的进一步封装，它大大的简化了多线程开发。concurrent包分成了三个部分，分别是java.util.concurrent、java.util.concurrent.atomic和java.util.concurrent.lock。内容涵盖了并发集合类、线程池机制、同步互斥机制、线程安全的变量更新工具类、锁等等常用工具。

       Executor：具体Runnable任务的执行者。

ExecutorService:一个线程池管理者，其实现类有多种，比如普通线程池，定时调度线程池ScheduledExecutorService等，我们能把一个

Runnable,Callable提交到池中让其调度。

Future：是与Runnable,Callable进行交互的接口，比如一个线程执行结束后取返回的结果等等，还提供了cancel终止线程。

BlockingQueue：阻塞队列。

 

 

 

二、一个任务队列的BlockingQueue实现

 

public int FetchInvertal = 1000;
public Datum newdatum = null;
// Creates a LinkedBlockingQueue with a capacity of Integer.MAX_VALUE.
// 理论上Integer.MAX_VALUE.的任务排队
final BlockingQueue<Datum> queue = new LinkedBlockingQueue<Datum>();
// 线程池
// 其中一个线程，取出待处理资料，put到queue
// 其余四个线程处理具体业务
final ExecutorService fetchdataservice = Executors
				.newSingleThreadExecutor();
final ExecutorService convertservice = Executors.newFixedThreadPool(4);
//计数
final AtomicInteger wc = new AtomicInteger();
final AtomicReference<String> atomstarttime=new AtomicReference<String>(null);
// FetchList线程
Runnable fetchlist = new Runnable() {
public void run() {
while (true) {
	try {
		List<Datum> tmpls = getLatestDatums("1", tomstarttime.get());
		if (tmpls != null && tmpls.size() > 0) {
		queue.addAll(tmpls);
		atomstarttime.set(tmpls.get(tmpls.size() - 1).getAddtime());
		tmpls=null;
		}
// 每隔一秒钟就检测一下是否有新的待处理的数据
		Thread.sleep(FetchInvertal);
		} catch (InterruptedException ex) {
	}
}
}
};
fetchdataservice.submit(fetchlist);
fetchdataservice.shutdown();

// 四个处理线程
for (int index = 0; index < 4; index++) {
		Runnable exewrite = new Runnable() {
		public void run() {
		int port = 8100 + wc.getAndIncrement();
		while (true) {
		try {
//如果队列里没有数据，会自动退出当前进程，为了防止进程被停掉，先判断是否有任务队列 
//队列里去看有没有任务，一直循环。
			if (queue.size() > 0) {
					Datum datum = queue.take();
					if (datum != null) {
					if (dobiz==true) {//此条件是伪码
										// 若成功，自动审核通过
						updateDatumState(datum.getUuid(), "3");
					} else {
						// 若不成功，自动审核不通过
						updateDatumState(datum.getUuid(), "2");
					}
				}
			} else {
					Thread.sleep(2000);//如果队列里没有任务了，睡眠两秒
			}
		} catch (InterruptedException e) {
	}
}
}
};
convertservice.submit(exewrite);
}
convertservice.shutdown();
}

private List<Datum> getLatestDatums(String status, String starttime) {
		//返回最新产生的待处理的任务列表
	}

private String updateDatumState(String uuid, String status) {
		//更新当前任务状态，打上已处理完成，或处理异常的标志
	}