ThreadPool,ObjectPool和任务控制

New a threadPool

ThreadPoolExecutor executorService = new ThreadPoolExecutor( poolCoreSize, poolMaxSize, timeout, java.util.concurrent.TimeUnit.MILLISECONDS, buff, handler);

 

poolcoreSize:  Pool至少运行的线程数量

poolMaxSize:   Pool最多运行的线程数量

timeout:空闲线程发呆时间,如果某线程发呆时间>timeout 而当前线程数量>poolcoreSize,就会终止该线程

buff:线程池所用的缓冲队列 实现BlockingQueue的那几个都能用,能设置定长和无限长度

handler: 当线程池线程数到达poolMaxSize buff 又满了,有新的任务丢到线程池的时候就会call这个hander进行处理.默认提供了几种处理方式,也可以自己实现,例如

public class PoolRejectHandler implements RejectedExecutionHandler { /* (non-Javadoc) * @see java.util.concurrent.RejectedExecutionHandler#rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) */ public void rejectedExecution(Runnable arg0, ThreadPoolExecutor arg1) { //log process is slower then produce //put back to inqueue BaseTask task = (BaseTask)arg0; //task.pushToInQueue(task.getInObject()); System.out.println("!!!!!!overload!!!"); try { task.getInQueue().put(task.getInObject()); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }

 这里我把这个task丢回一开始的任务队列了,这里可以实现自己的处理逻辑

 

New and excute  task

任何实现runnable 或者callable接口的类 都可以作为一个任务丢到线程池里面运行

executorService.execute(processTask);

 

ThreadPool 调度策略

条件: 线程数<poolcoreSize, buff未满,有新的任务提交   结果:新建线程处理这个任务

条件: 线程数>=poolcoreSize, buff未满,有新的任务提交  结果:把任务丢到buff对列中

条件: poolMaxSize>线程数>=poolcoreSize, buff满,有新的任务提交  结果:新建线程

条件: poolMaxSize=线程数>poolcoreSize, buff满,有新的任务提交  结果:调用handler处理这个任务

 

 

这里需要讨论的就是最后那个情况,虽然threadpool提供了handler来处理这种超出负荷的情况,但是如果能从源头就直接控制提交到线程池的任务数量,感觉上会好很多

 

 

这里我用到一个Manager类和ObjectPool来处理任务的提交

 

Manager类实现Runnable 和 DispatchListener 接口 ,在Runnable里面处理任务生成和分发,DispatchListener接口中实现任务对象的回收和结果收集(示意代码)

public class DispathManager implements Runnable, DispatchListener { public void run() { while (true) { try { // get task from taskobjectpool. will wait here if the active taskobjects reach the tasklimit // so we can control the limit of running task in the threadpool BaseTask processTask = (BaseTask) taskPool.borrowObject(); //set task params processTask.setListener(this); // put in threadpool executorService.execute(processTask); } catch (InterruptedException e) { break; } catch (Exception e) { // log dispatch queue exception e.printStackTrace(); } } } public void recycleTask(BaseTask baseTask) { // TODO Auto-generated method stub try { taskPool.returnObject(baseTask); } catch (Exception e) { e.printStackTrace(); } } }

 

然后在Task执行完毕的时候进行回收

public void run() { try { //task logic here // return task listener.recycleTask(this); } catch (Exception e) { listener.onTaskError(e, inObject); } }

 

这样就不必每次都重新new 这个task 对一些重型的task,会提高性能

 

这里用到了ObjectPool,是Apache Common下的管理ObjectPool的类,使用如下:

 

 

New a ObjectPool

ObjectPool taskPool = new GenericObjectPool(new TaskPoolFactory(taskClass), taskLimit, GenericObjectPool.WHEN_EXHAUSTED_BLOCK, -1);

 

其中TaskPoolFactory如下 taskClass是重配置文件读入的

public class TaskPoolFactory extends BasePoolableObjectFactory { /* (non-Javadoc) * @see org.apache.commons.pool.BasePoolableObjectFactory#makeObject() */ private Class c; public TaskPoolFactory(Class c) { super(); this.c = c; } @Override public Object makeObject() throws Exception { // TODO Auto-generated method stub return c.newInstance(); } }

 

taskLimit是允许借出的最大的数目(借出但是没有归还)

当借出的数目已经达到最大，这个时候还有借出请求 GenericObjectPool.WHEN_EXHAUSTED_BLOCK 就提供了策略（还有几种的) 这里设置为无限等待，也就是线程会wait在这里 直到有借出的task被归还

 

Objectpool使用参照Manager和task类

 

最后, 控制taskLimit的大小 就可以控制ThreadPool中处理任务的多少，也就是正在处理的task+buff里的task 必定<= taskLimit,这样就在源头控制了提交任务的数量,保证threadpool不会超负荷。

 

 

最后推荐大家的设置就是　ThreadPool buff设置为无限大　然后用poolcoresize和taskLimit来控制负荷能力（视不同机器设置不同的值)

 

 

 

欢迎大家拍砖．．．