您的位置:首页 > 编程语言 > Java开发

Java 线程池

2016-03-25 16:10 295 查看
何时需要线程池

创建和销毁线程的时间很长

线程池一般在程序执行开始和结束阶段负责线程初始化和销毁,节约时间

请求数目非常多

如每个请求都需要一个线程

线程池统一缓冲,不至于并发太多

ThreadPoolExecutor构造函数参数

构造器中各个参数含义:

corePoolSize

核心池的大小。

默认情况下,在创建了线程池后,线程池中的线程数为0,当有任务来之后,就会创建一个线程去执行任务,当线程池中的线程数目达到corePoolSize后,就会把到达的任务放到缓存队列当中;

maximumPoolSize

线程池最大线程数,这个参数也是一个非常重要的参数,它表示在线程池中最多能创建多少个线程;

keepAliveTime

表示线程没有任务执行时最多保持多久时间会终止。

默认情况下,只有当线程池中的线程数大于corePoolSize时,keepAliveTime才会起作用,直到线程池中的线程数不大于corePoolSize,即当线程池中的线程数大于corePoolSize时,如果一个线程空闲的时间达到keepAliveTime,则会终止,直到线程池中的线程数不超过corePoolSize。

但是如果调用了allowCoreThreadTimeOut(boolean)方法,在线程池中的线程数不大于corePoolSize时,keepAliveTime参数也会起作用,直到线程池中的线程数为0。

keepAliveTime和maximumPoolSize及BlockingQueue的类型均有关系。

如果BlockingQueue是无界的,那么永远不会触发maximumPoolSize,自然keepAliveTime也就没有了意义。

反之,如果核心数较小,有界BlockingQueue数值又较小,同时keepAliveTime又设的很小,如果任务频繁,那么系统就会频繁的申请回收线程。

unit

参数keepAliveTime的时间单位,有7种取值,在TimeUnit类中有7种静态属性:

TimeUnit.DAYS; //天

TimeUnit.HOURS; //小时

TimeUnit.MINUTES; //分钟

TimeUnit.SECONDS; //秒

TimeUnit.MILLISECONDS; //毫秒

TimeUnit.MICROSECONDS; //微妙

TimeUnit.NANOSECONDS; //纳秒

workQueue

一个阻塞队列,用来存储等待执行的任务,一般使用LinkedBlockingQueue和Synchronous

threadFactory

线程工厂,主要用来创建线程

handler

表示当拒绝处理任务时的策略,有以下四种取值:

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。

ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。

ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)

ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务

线程池运行流程

1、线程池刚创建时,里面没有一个线程。任务队列是作为参数传进来的。不过,就算队列里面有任务,线程池也不会马上执行它们。

2、当调用 execute() 方法添加一个任务时,线程池会做如下判断:

a. 如果正在运行的线程数量小于 corePoolSize,那么马上创建线程运行这个任务;

b. 如果正在运行的线程数量大于或等于 corePoolSize,那么将这个任务放入队列。

c. 如果这时候队列满了,而且正在运行的线程数量小于 maximumPoolSize,那么还是要创建线程运行这个任务;

d. 如果队列满了,而且正在运行的线程数量大于或等于 maximumPoolSize,那么线程池会抛出异常,告诉调用者“我不能再接受任务了”。

4、当一个线程无事可做,超过一定的时间(keepAliveTime)时,线程池会判断,如果当前运行的线程数大于 corePoolSize,那么这个线程就被停掉。所以线程池的所有任务完成后,它最终会收缩到 corePoolSize 的大小。

这个过程说明,并不是先加入任务就一定会先执行。假设队列大小为 4,corePoolSize为2,maximumPoolSize为6,那么当加入15个任务时,执行的顺序类似这样:首先执行任务 1、2,然后任务3~6被放入队列。

这时候队列满了,任务7、8、9、10 会被马上执行,而任务 11~15 则会抛出异常。最终顺序是:1、2、7、8、9、10、3、4、5、6。当然这个过程是针对指定大小的
ArrayBlockingQueue<Runnable>
来说,如果是
LinkedBlockingQueue<Runnable>
,因为该队列无大小限制,所以不存在上述问题。

只有ArrayBlockingQueue才有上面这个情况:

public class ThreadPoolTest implements Runnable {
private int ino;

ThreadPoolTest (int i) {
ino = i;
}

public void run() {
synchronized (this) {
try {
//                System.out.println(Thread.currentThread().getName());
System.out.println("task no: " + ino);
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/*
* 并不是先加入任务就一定会先执行。
* 假设队列大小为 4,corePoolSize为2,maximumPoolSize为6.
* 那么当加入15个任务时,执行的顺序类似这样:
* 首先执行任务 1、2,
* 然后任务3~6被放入队列。
* 这时候队列满了,任务7、8、9、10 会被马上执行,而任务 11~15 则会抛出异常。
* 最终顺序是:1、2、7、8、9、10、3、4、5、6。
* 当然这个过程是针对指定大小的ArrayBlockingQueue<Runnable>来说,
* 如果是LinkedBlockingQueue<Runnable>,因为该队列无大小限制,所以不存在上述问题。
*/
public static void main(String[] args) {
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(5);
ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
for (int i = 0; i < 15; i++) {
executor.execute(new Thread(new ThreadPoolTest(i), "TestThread".concat("" + i)));
int threadSize = queue.size();
System.out.println("线程队列大小为-->" + threadSize);
}
//        executor.shutdown();
}
}


Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task Thread[TestThread11,5,main] rejected from java.util.concurrent.ThreadPoolExecutor@2c7b84de[Running, pool size = 6, active threads = 6, queued tasks = 5, completed tasks = 0]


很明显,抛RejectedExecutionException异常了,被拒绝策略拒绝了,这就说明线程超出了线程池的总容量(线程队列大小+最大线程数)。

但是在BlockQueue满时可以通过put函数重试添加进队列:

public static void main(String[] args) {
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(5);
ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
for (int i = 0; i < 15; i++) {
executor.execute(new Thread(new ThreadPoolTest(i), "TestThread".concat("" + i)));
int threadSize = queue.size();
System.out.println("线程队列大小为-->" + threadSize);
try {
//线程满时,通过put依然可以将新线程添加进队列
if (threadSize==4) {
queue.put(new Runnable() {
@Override
public void run(){
System.out.println("我是新线程,看看能不能搭个车加进去!");
}
});
}
} catch (InterruptedException e) {
e.printStackTrace();
}

}
//        executor.shutdown();
}


效果:

task no: 2
task no: 3
task no: 4
task no: 5
我是新线程,看看能不能搭个车加进去!


线程池的排除策略

A. 如果运行的线程少于 corePoolSize,则 Executor 始终首选添加新的线程,而不进行排队。

B. 如果运行的线程等于或多于 corePoolSize,则 Executor 始终首选将请求加入队列,而不添加新的线程。

C. 如果无法将请求加入队列,则创建新的线程,除非创建此线程超出 maximumPoolSize,在这种情况下,任务将被拒绝。

线程池的拒绝策略

RejectedExecutionHandler

即使向老板借了工人,但是任务还是继续过来,还是忙不过来,这时整个队伍只好拒绝接受了。

RejectedExecutionHandler接口提供了对于拒绝任务的处理的自定方法的机会。

在ThreadPoolExecutor中已经默认包含了4中策略,因为源码非常简单,这里直接贴出来。

CallerRunsPolicy:线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制,能够减缓新任务的提交速度。

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}


这个策略显然不想放弃执行任务。但是由于池中已经没有任何资源了,那么就直接使用调用该execute的线程本身来执行。

AbortPolicy:处理程序遭到拒绝将抛出运行时RejectedExecutionException

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException();
}


这种策略直接抛出异常,丢弃任务。

DiscardPolicy:不能执行的任务将被删除

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}


这种策略和AbortPolicy几乎一样,也是丢弃任务,只不过他不抛出异常。

DiscardOldestPolicy:如果执行程序尚未关闭,则位于工作队列头部的任务将被删除,然后重试执行程序(如果再次失败,则重复此过程)

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}


该策略就稍微复杂一些,在pool没有关闭的前提下首先丢掉缓存在队列中的最早的任务,然后重新尝试运行该任务。这个策略需要适当小心。

设想:如果其他线程都还在运行,那么新来任务踢掉旧任务,缓存在queue中,再来一个任务又会踢掉queue中最老任务。

ThreadPoolExecutor的方法



Executor

顶层接口,在它里面只声明了一个方法execute(Runnable),返回值为void,参数为Runnable类型,用来执行传进去的任务的;

ExecutorService

继承了Executor接口,并声明了一些方法:submit、invokeAll、invokeAny以及shutDown等;

AbstractExecutorService

实现了ExecutorService接口,基本实现了ExecutorService中声明的所有方法;

ThreadPoolExecutor

继承了类AbstractExecutorService。

在ThreadPoolExecutor类中有几个非常重要的方法:

execute()
submit()
shutdown()
shutdownNow()


execute()方法实际上是Executor中声明的方法,在ThreadPoolExecutor进行了具体的实现,这个方法是ThreadPoolExecutor的核心方法,通过这个方法可以向线程池提交一个任务,交由线程池去执行。

submit()方法是在ExecutorService中声明的方法,在AbstractExecutorService就已经有了具体的实现,在ThreadPoolExecutor中并没有对其进行重写,这个方法也是用来向线程池提交任务的,但是它和execute()方法不同,它能够返回任务执行的结果,去看submit()方法的实现,会发现它实际上还是调用的execute()方法,只不过它利用了Future来获取任务执行结果。

shutdown()和shutdownNow()是用来关闭线程池的

线程池工具类

利用Executors创建线程池:

newCachedThreadPool创建一个可缓存线程池,如果线程池长度超过处理需要,可灵活回收空闲线程,若无可回收,则新建线程。

newFixedThreadPool 创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待。

newScheduledThreadPool 创建一个定长线程池,支持定时及周期性任务执行。

newSingleThreadExecutor 创建一个单线程化的线程池,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

newCachedThreadPool

创建一个可缓存线程池,如果线程池长度超过处理需要,可灵活回收空闲线程,若无可回收,则新建线程。示例代码如下:

package test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExecutorTest {
public static void main(String[] args) {
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int index = i;
try {
Thread.sleep(index * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
cachedThreadPool.execute(new Runnable() {
public void run() {
System.out.println(index);
}
});
}
}
}


线程池为无限大,当执行第二个任务时第一个任务已经完成,会复用执行第一个任务的线程,而不用每次新建线程。

newFixedThreadPool

创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待。示例代码如下:

package test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExecutorTest {
public static void main(String[] args) {
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
for (int i = 0; i < 10; i++) {
final int index = i;
fixedThreadPool.execute(new Runnable() {
public void run() {
try {
System.out.println(index);
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
}


因为线程池大小为3,每个任务输出index后sleep 2秒,所以每两秒打印3个数字。

定长线程池的大小最好根据系统资源进行设置。如
Runtime.getRuntime().availableProcessors()


newScheduledThreadPool

创建一个定长线程池,支持定时及周期性任务执行。延迟执行示例代码如下:

package test;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExecutorTest {
public static void main(String[] args) {
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
scheduledThreadPool.schedule(new Runnable() {
public void run() {
System.out.println("delay 3 seconds");
}
}, 3, TimeUnit.SECONDS);
}
}


表示延迟3秒执行。

定期执行示例代码如下:

package test;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExecutorTest {
public static void main(String[] args) {
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
public void run() {
System.out.println("delay 1 seconds, and excute every 3 seconds");
}
}, 1, 3, TimeUnit.SECONDS);
}
}


表示延迟1秒后每3秒执行一次。

newSingleThreadExecutor

创建一个单线程化的线程池,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。示例代码如下:

package test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExecutorTest {
public static void main(String[] args) {
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
final int index = i;
singleThreadExecutor.execute(new Runnable() {
public void run() {
try {
System.out.println(index);
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
}


结果依次输出,相当于顺序执行各个任务。

可以使用JDK自带的监控工具来监控我们创建的线程数量,运行一个不终止的线程,创建指定量的线程,来观察:

package test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExecutorTest {
public static void main(String[] args) {
ExecutorService singleThreadExecutor = Executors.newCachedThreadPool();
for (int i = 0; i < 100; i++) {
final int index = i;
singleThreadExecutor.execute(new Runnable() {
public void run() {
try {
while (true) {
System.out.println(index);
Thread.sleep(10 * 1000);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}






总结

1)首先,要清楚corePoolSize和maximumPoolSize的含义;

2)其次,要知道Worker是用来起到什么作用的;

3)要知道任务提交给线程池之后的处理策略,这里总结一下主要有4点:

如果当前线程池中的线程数目小于corePoolSize,则每来一个任务,就会创建一个线程去执行这个任务;

如果当前线程池中的线程数目>=corePoolSize,则每来一个任务,会尝试将其添加到任务缓存队列当中,若添加成功,则该任务会等待空闲线程将其取出去执行;若添加失败(一般来说是任务缓存队列已满),则会尝试创建新的线程去执行这个任务;

如果当前线程池中的线程数目达到maximumPoolSize,则会采取任务拒绝策略进行处理;

如果线程池中的线程数量大于 corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止,直至线程池中的线程数目不大于corePoolSize;如果允许为核心池中的线程设置存活时间,那么核心池中的线程空闲时间超过keepAliveTime,线程也会被终止。

参考链接

JAVA线程池ThreadPoolExecutor与阻塞队列BlockingQueue

Java8并发教程:Threads和Executors

Java线程池使用说明

Java并发编程:线程池的使用
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: