用JAVA写的一个异步多线程批处理的组件

package util; import java.util.LinkedList; import java.util.List; public class JobDispatcher extends Thread { private List<JobEvent> mJobList = new LinkedList<JobEvent>(); private List<JobWorker> mIdleWorkers = new LinkedList<JobWorker>(); public static int MAX_WORKER_NUM = 5; public static int SLEEP_TIME = 1000; private boolean mCanRetire; public JobDispatcher() { this.mCanRetire = false; } public synchronized void addIdleWorker(JobWorker worker) { if(worker == null) { return; } if(this.getIdleWorkerNum() < JobDispatcher.MAX_WORKER_NUM) { this.mIdleWorkers.add(worker); if(worker.getState() == Thread.State.NEW) { worker.start(); } } } public synchronized void addJob(JobEvent job) { if(this.mJobList == null) { return; } else { this.mJobList.add(job); } } private synchronized int getIdleWorkerNum() { return this.mIdleWorkers.size(); } private synchronized JobWorker findFreeWorker() { for(int i = 0; i < this.getIdleWorkerNum(); i++) { JobWorker worker = this.mIdleWorkers.remove(i); if(worker != null) { return worker; } } return null; } private synchronized boolean isJobListEmpty() { return this.mJobList.isEmpty(); } private synchronized JobEvent getTopJob() { if(this.mJobList == null || this.mJobList.size() == 0) { return null; } else { return this.mJobList.remove(0); } } public synchronized boolean CanRetire() { return mCanRetire; } public synchronized void setCanRetire(boolean mCanRetire) { this.mCanRetire = mCanRetire; } public void run() { // 检查 JobList 中是否有工作要做， while(!this.CanRetire()) { if(!this.isJobListEmpty()) { // 有工作要做，找闲置工人 JobWorker freeWorker = this.findFreeWorker(); if(freeWorker != null) { // 有工作，并且当前有闲置的工人 JobEvent job = this.getTopJob(); if(job == null) { System.out.println("不知什么原因没有工作了。。。"); } else { // 把工作分配给这个工人做 freeWorker.setJob(job); } } else { // 有工作，但当前没有闲置的工人，则先暂时搁置这个工作，等待闲置工人然后再分配给他做 try { Thread.sleep(JobDispatcher.SLEEP_TIME); } catch (Exception e) { System.out.println("有工作但没有闲置工人，等待闲置工人时错误" + e.getMessage()); } } } else { // 没有工作则继续等待加入工作 try { Thread.sleep(JobDispatcher.SLEEP_TIME); } catch (Exception e) { System.out.println("没有工作，等待加入工作时错误" + e.getMessage()); } } } // retire 所有工人 for(int i = 0; i < JobDispatcher.MAX_WORKER_NUM; i++) { JobWorker worker = this.mIdleWorkers.get(i); if(worker != null) { worker.setCanRetire(true); } } } public boolean TimeToRetireAll() { System.out.println("待做工作数["+ this.mJobList.size() +"]"); System.out.println("闲置工人数["+ this.getIdleWorkerNum() +"]"); if(this.mJobList == null || this.isJobListEmpty()) { if(this.getIdleWorkerNum() == MAX_WORKER_NUM) { return true; } else { return false; } } else { return false; } } }