线程池ThreadPoolExecutor使用简介

时间 2020-02-12

标签线程 threadpoolexecutor 使用简介繁體版

原文原文链接

1、简介
线程池类为 java.util.concurrent.ThreadPoolExecutor，经常使用构造方法为：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue workQueue,
RejectedExecutionHandler handler)
corePoolSize：线程池维护线程的最少数量
maximumPoolSize：线程池维护线程的最大数量
keepAliveTime：线程池维护线程所容许的空闲时间
unit：线程池维护线程所容许的空闲时间的单位
workQueue：线程池所使用的缓冲队列
handler：线程池对拒绝任务的处理策略

一个任务经过 execute(Runnable)方法被添加到线程池，任务就是一个 Runnable类型的对象，任务的执行方法就是 Runnable类型对象的run()方法。

当一个任务经过execute(Runnable)方法欲添加到线程池时：

若是此时线程池中的数量小于corePoolSize，即便线程池中的线程都处于空闲状态，也要建立新的线程来处理被添加的任务。
若是此时线程池中的数量等于 corePoolSize，可是缓冲队列 workQueue未满，那么任务被放入缓冲队列。
若是此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，而且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。
若是此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，而且线程池中的数量等于maximumPoolSize，那么经过 handler所指定的策略来处理此任务。

也就是：处理任务的优先级为：
核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize，若是三者都满了，使用handler处理被拒绝的任务。

当线程池中的线程数量大于 corePoolSize时，若是某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池能够动态的调整池中的线程数。

unit可选的参数为java.util.concurrent.TimeUnit中的几个静态属性：
NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS。

workQueue我经常使用的是：java.util.concurrent.ArrayBlockingQueue

handler有四个选择：
ThreadPoolExecutor.AbortPolicy()
抛出java.util.concurrent.RejectedExecutionException异常
ThreadPoolExecutor.CallerRunsPolicy()
重试添加当前的任务，他会自动重复调用execute()方法
ThreadPoolExecutor.DiscardOldestPolicy()
抛弃旧的任务
ThreadPoolExecutor.DiscardPolicy()
抛弃当前的任务

2、通常用法举例

点击(此处)折叠或打开 java

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;
}
}

说明：
一、在这段程序中，一个任务就是一个Runnable类型的对象，也就是一个ThreadPoolTask类型的对象。
二、通常来讲任务除了处理方式外，还须要处理的数据，处理的数据经过构造方法传给任务。
三、在这段程序中，main()方法至关于一个残忍的领导，他派发出许多任务，丢给一个叫 threadPool的不辞辛苦的小组来作。
这个小组里面队员至少有两个，若是他们两个忙不过来，任务就被放到任务列表里面。
若是积压的任务过多，多到任务列表都装不下(超过3个)的时候，就雇佣新的队员来帮忙。可是基于成本的考虑，不能雇佣太多的队员，至多只能雇佣 4个。
若是四个队员都在忙时，再有新的任务，这个小组就处理不了了，任务就会被经过一种策略来处理，咱们的处理方式是不停的派发，直到接受这个任务为止(更残忍！呵呵)。
由于队员工做是须要成本的，若是工做很闲，闲到 3SECONDS都没有新的任务了，那么有的队员就会被解雇了，可是，为了小组的正常运转，即便工做再闲，小组的队员也不能少于两个。
四、经过调整 produceTaskSleepTime和 consumeTaskSleepTime的大小来实现对派发任务和处理任务的速度的控制，改变这两个值就能够观察不一样速率下程序的工做状况。
五、经过调整4中所指的数据，再加上调整任务丢弃策略，换上其余三种策略，就能够看出不一样策略下的不一样处理方式。
六、对于其余的使用方法，参看jdk的帮助，很容易理解和使用。

另外一个例子:

点击(此处)折叠或打开 this

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();
}
}
}
}