一、简介
线程池类为 java.util.concurrent.ThreadPoolExecutor,常用构造方法为:
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, RejectedExecutionHandler handler)
一个任务通过 execute(Runnable)方法被添加到线程池,任务就是一个 Runnable类型的对象,任务的执行方法就是 Runnable类型对象的run()方法。
当一个任务通过execute(Runnable)方法欲添加到线程池时:
也就是:处理任务的优先级为:
核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize,如果三者都满了,使用handler处理被拒绝的任务。
当线程池中的线程数量大于 corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止。这样,线程池可以动态的调整池中的线程数。
unit可选的参数为java.util.concurrent.TimeUnit中的几个静态属性:
NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS。
workQueue我常用的是:java.util.concurrent.ArrayBlockingQueue
handler有四个选择:
二、一般用法举例
package demo; import java.io.Serializable; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; public class TestThreadPool2 { private static int produceTaskSleepTime = 2; private static int produceTaskMaxNumber = 10; public static void main(String[] args) { // 构造一个线程池 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3), new ThreadPoolExecutor.DiscardOldestPolicy()); for (int i = 1; i <= produceTaskMaxNumber; i++) { try { // 产生一个任务,并将其加入到线程池 String task = "task@ " + i; System.out.println("put " + task); threadPool.execute(new ThreadPoolTask(task)); // 便于观察,等待一段时间 Thread.sleep(produceTaskSleepTime); } catch (Exception e) { e.printStackTrace(); } } } } /** * 线程池执行的任务 */ class ThreadPoolTask implements Runnable, Serializable { private static final long serialVersionUID = 0; private static int consumeTaskSleepTime = 2000; // 保存任务所需要的数据 private Object threadPoolTaskData; ThreadPoolTask(Object tasks) { this.threadPoolTaskData = tasks; } public void run() { // 处理一个任务,这里的处理方式太简单了,仅仅是一个打印语句 System.out.println(Thread.currentThread().getName()); System.out.println("start .." + threadPoolTaskData); try { // //便于观察,等待一段时间 Thread.sleep(consumeTaskSleepTime); } catch (Exception e) { e.printStackTrace(); } threadPoolTaskData = null; } public Object getTask() { return this.threadPoolTaskData; } }
说明:
1、在这段程序中,一个任务就是一个Runnable类型的对象,也就是一个ThreadPoolTask类型的对象。
2、一般来说任务除了处理方式外,还需要处理的数据,处理的数据通过构造方法传给任务。
3、在这段程序中,main()方法相当于一个残忍的领导,他派发出许多任务,丢给一个叫 threadPool的任劳任怨的小组来做。
这个小组里面队员至少有两个,如果他们两个忙不过来,任务就被放到任务列表里面。
如果积压的任务过多,多到任务列表都装不下(超过3个)的时候,就雇佣新的队员来帮忙。但是基于成本的考虑,不能雇佣太多的队员,至多只能雇佣 4个。
如果四个队员都在忙时,再有新的任务,这个小组就处理不了了,任务就会被通过一种策略来处理,我们的处理方式是不停的派发,直到接受这个任务为止(更残忍!呵呵)。
因为队员工作是需要成本的,如果工作很闲,闲到 3SECONDS都没有新的任务了,那么有的队员就会被解雇了,但是,为了小组的正常运转,即使工作再闲,小组的队员也不能少于两个。
4、通过调整 produceTaskSleepTime和 consumeTaskSleepTime的大小来实现对派发任务和处理任务的速度的控制,改变这两个值就可以观察不同速率下程序的工作情况。
5、通过调整4中所指的数据,再加上调整任务丢弃策略,换上其他三种策略,就可以看出不同策略下的不同处理方式。
6、对于其他的使用方法,参看jdk的帮助,很容易理解和使用。
另一个例子:
package demo; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; public class ThreadPoolExecutorTest { private static int queueDeep = 4; public void createThreadPool() { /* * 创建线程池,最小线程数为2,最大线程数为4,线程池维护线程的空闲时间为3秒, * 使用队列深度为4的有界队列,如果执行程序尚未关闭,则位于工作队列头部的任务将被删除, * 然后重试执行程序(如果再次失败,则重复此过程),里面已经根据队列深度对任务加载进行了控制。 */ ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueDeep), new ThreadPoolExecutor.DiscardOldestPolicy()); // 向线程池中添加 10 个任务 for (int i = 0; i < 10; i++) { try { Thread.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } while (getQueueSize(tpe.getQueue()) >= queueDeep) { System.out.println("队列已满,等3秒再添加任务"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } TaskThreadPool ttp = new TaskThreadPool(i); System.out.println("put i:" + i); tpe.execute(ttp); } tpe.shutdown(); } private synchronized int getQueueSize(Queue queue) { return queue.size(); } public static void main(String[] args) { ThreadPoolExecutorTest test = new ThreadPoolExecutorTest(); test.createThreadPool(); } class TaskThreadPool implements Runnable { private int index; public TaskThreadPool(int index) { this.index = index; } public void run() { System.out.println(Thread.currentThread() + " index:" + index); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } } }
总结
以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,谢谢大家对呐喊教程的支持。如果你想了解更多相关内容请查看下面相关链接
声明:本文内容来源于网络,版权归原作者所有,内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:notice#nhooo.com(发邮件时,请将#更换为@)进行举报,并提供相关证据,一经查实,本站将立刻删除涉嫌侵权内容。