java阻塞线程池_线程池解决阻塞方法
一、序言
当我们需要使用线程的时候,我们可以新建一个线程,然后显式调用线程的start()方法,这样实现起来非常简便,但在某些场景下存在缺陷:如果需要同时执行多个任务(即并发的线程数量很多),频繁地创建线程会降低系统的效率,因为创建和销毁线程均需要一定的时间。
线程池可以使线程得到复用,所谓线程复用就是线程在执行完一个任务后并不被销毁,该线程可以继续执行其他的任务。java.lang.concurrent包中的Executors类为我们创建线程池提供了方便。
二、Executors的简单使用示例
此处我们先来看一个简单的例子,如下:
package com.soft;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ExecutorsDemo {
public static void main(String[] args) throws InterruptedException, ExecutionException {
// ExecutorService executor = Executors.newSingleThreadExecutor();
// ExecutorService executor = Executors.newCachedThreadPool();
ExecutorService executor = Executors.newFixedThreadPool(5);
Thread.sleep(5*1000);//方便监控工具能捕获到
for (int i = 0; i < 10; i++) {
final int no = i;
Runnable runnable = new Runnable() {
public void run() {
try {
System.out.println("into" + no);
Thread.sleep(1000L);
System.out.println("end" + no);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
executor.execute(runnable);//ExecutorService有一个execute()方法,这个方法的参数是Runnable类型,通过execute(Runnable)方法即可将一个任务添加到线程池,任务的执行方法是Runnable类型对象的run()方法。
}//End for
executor.shutdown();
System.out.println("Thread Main End!");
}
}
其运行结果如下:
into0
into3
Thread Main End!
into4
into1
into2
end0
into5
end3
end1
end4
into8
into6
into7
end2
into9
end5
end7
end8
end6
end9
解说:这个例子应该很容易看懂,从运行结果来看,在任意某一时刻只有5个线程在执行,这是因为上述代码通过Executors.newFixedThreadPool(5)语句创建了一个固定长度的线程池(长度为5),一个结束之后另再一个才开始执行。
三、Executors提供的线程池
Executors是线程的工厂类,也可以说是一个线程池工具类,它调用其内部静态方法(如newFixedThreadPool()等)即可创建一个线程池,通过参数设置,Executors提供不同的线程池机制。
四、简述线程池的属性
五、详解ThreadPoolExecutor
上文提到可以通过显式的ThreadPoolExecutor构造函数来构造特定形式的线程池,ThreadPoolExecutor是java.util.concurrent包以内部线程池的形式对外提供线程池管理、线程调度等服务,此处我们来了解一下ThreadPoolExecutor
(1)一般使用方式:
ExecutorService exec = new ThreadPoolExecutor(8,
8,
0L,
TimeUnit.MILLISECONDS,
new LinkedBlockingQueue(100),
new ThreadPoolExecutor.CallerRunsPolicy());
下文详解此示例涉及的一些内容
(2)构造函数的声明:
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
(3)函数参数说明:
参数名
代表含义
corePoolSize
线程池的基本大小(核心线程池大小)
maximumPoolSize
线程池的最大大小
keepAliveTime
线程池中超过corePoolSize数目的空闲线程的最大存活时间
unit
keepAliveTime参数的时间单位
workQueue
任务阻塞队列
threadFactory
新建线程的工厂
handler
当提交的任务数超过maxmumPoolSize与workQueue之和时,任务会交给RejectedExecutionHandler来处理
进一步解说:
A、当提交新任务时,若线程池大小小于corePoolSize,将创建一个新的线程来执行任务,即使此时线程池中存在空闲线程;
B、当提交新任务时,若线程池达到corePoolSize大小,新提交的任务将被放入workQueue中,等待线程池调度执行;
C、当提交新任务时,若workQueue已满,且maximumPoolSize>corePoolSize,将创建新的线程来执行任务;
D、当提交新任务时,若任务总数超过maximumPoolSize,新提交的任务将由RejectedExecutionHandler来处理;
E、当线程池中的线程数超过corePoolSize时,若线程的空闲时间达到keepAliveTime,则关闭空闲线程
(4)任务阻塞队列选择机制
(5)简述SynchronousQueue
注:此处贴出SynchronousQueue的使用示例,示例中使用了Semaphore,更多关于SynchronousQueue及Semaphore的内容请参考其他文章
package com.test;
import java.util.concurrent.Semaphore;
import java.util.concurrent.SynchronousQueue;
/*
* 程序中有10个线程来消费生成者产生的数据,这些消费者都调用TestDo.doSome()方法去进行处理,
* 每个消费者都需要一秒才能处理完,程序应保证这些消费者线程依次有序地消费数据,只有上一个消费者消费完后,
* 下一个消费者才能消费数据,下一个消费者是谁都可以,但要保证这些消费者线程拿到的数据是有顺序的。
*/
public class SynchronousQueueTest {
public static void main(String[] args) {
System.out.println("begin:" + (System.currentTimeMillis() / 1000));
// 定义一个Synchronous
final SynchronousQueue sq = new SynchronousQueue();
// 定义一个数量为1的信号量,其作用相当于一个互斥锁
final Semaphore sem = new Semaphore(1);
for (int i = 0; i < 10; i++) {
new Thread(new Runnable() {
public void run() {
try {
sem.acquire();
String input = sq.take();
String output = TestDo.doSome(input);//内部类
System.out.println(Thread.currentThread().getName()+ ":" + output);
sem.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
for (int i = 0; i < 10; i++) {
String input = i + ""; //此处将i变成字符串
try {
sq.put(input);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}//End main
}
class TestDo {
public static String doSome(String input) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
String output = input + ":" + (System.currentTimeMillis() / 1000);
return output;
}
}
上述代码的运行结果如下:
begin:1458954798
Thread-0:0:1458954799
Thread-1:1:1458954800
Thread-2:2:1458954801
Thread-3:3:1458954802
Thread-4:4:1458954803
Thread-5:5:1458954804
Thread-6:6:1458954805
Thread-7:7:1458954806
Thread-8:8:1458954807
Thread-9:9:1458954808
从上述结果看,上例在任意某一时刻只有一个线程在执行,且只有前一个线程执行完下一个线程才开始
六、饱和策略(线程池任务拒绝策略)
上文提到ThreadPoolExecutor构造函数的RejectedExecutionHandler handler参数,该参数表示当提交的任务数超过maxmumPoolSize与workQueue之和时,任务会交给RejectedExecutionHandler来处理,此处我们来具体了解一下
(1)四种饱和策略
(2)源码分析:
RejectedExecutionHandler这个接口是用来处理被丢弃的线程的异常处理接口,其源码如下:
public interface RejectedExecutionHandler{
//被线程池丢弃的线程处理机制
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) ;
}
AbortPolicy(中止策略)继承RejectedExecutionHandler接口,其源码如下:
public static class AbortPolicy implements RejectedExecutionHandler{
public AbortPolicy(){}
//直接抛出异常
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
throw new RejectedExecutionException("Task"+r.toString()+"rejected from"+executor.toString());
}
}
我们可以自己实现RejectedExecutionHandler接口,将实现类作为线程丢弃处理类,代码如下:
package com.test;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
public class RejectedExecutionHandlerDemo implements RejectedExecutionHandler{
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
// TODO Auto-generated method stub
System.out.println("线程信息"+r.toString()+"被遗弃的线程池:"+executor.toString());
}
}
七、定制ThreadPoolExecutor
(1)通过修改参数的方式达到定制目的
(2)通过自定义方式(封装各种参数)达到定制目的
示例(摘自网络):
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class CustomThreadPoolExecutor {
private ThreadPoolExecutor pool = null;
/**
* 线程池初始化方法
*
* corePoolSize 核心线程池大小----10
* maximumPoolSize 最大线程池大小----30
* keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间----30+单位TimeUnit
* TimeUnit keepAliveTime时间单位----TimeUnit.MINUTES
* workQueue 阻塞队列----new ArrayBlockingQueue(10)====10容量的阻塞队列
* threadFactory 新建线程工厂----new CustomThreadFactory()====定制的线程工厂
* rejectedExecutionHandler 当提交任务数超过maxmumPoolSize+workQueue之和时,
* 即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)),
* 任务会交给RejectedExecutionHandler来处理
*/
public void init() {
pool = new ThreadPoolExecutor(
10,
30,
30,
TimeUnit.MINUTES,
new ArrayBlockingQueue(10),
new CustomThreadFactory(),new CustomRejectedExecutionHandler());
}
public void destory() {
if(pool != null) {
pool.shutdownNow();
}
}
public ExecutorService getCustomThreadPoolExecutor() {
return this.pool;
}
private class CustomThreadFactory implements ThreadFactory {
private AtomicInteger count = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);
System.out.println(threadName);
t.setName(threadName);
return t;
}
}
private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
// 记录异常
// 报警处理等
System.out.println("error.............");
}
}
// 测试构造的线程池
public static void main(String[] args) {
CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor();
// 1.初始化
exec.init();
ExecutorService pool = exec.getCustomThreadPoolExecutor();
for(int i=1; i<100; i++) {
System.out.println("提交第" + i + "个任务!");
pool.execute(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("running=====");
}
});
}
// 2.销毁----此处不能销毁,因为任务没有提交执行完,如果销毁线程池,任务也就无法执行了
// exec.destory();
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
我们可以看到上述代码是通过init()方法对ThreadPoolExecutor构造函数进行了一些自定义设置,getCustomThreadPoolExecutor()方法返回init()方法配置的ThreadPoolExecutor对象实例(即线程池引用)
补充:
ThreadPoolExecutor构造函数有一个参数ThreadFactory threadFactory,前文提到该参数是新建线程的工厂,此处进一步解说该参数。
ThreadFactory是java.util.concurrent包下创建线程工厂的接口,ThreadFactory接口源码如下:
public interface ThreadFactory {
Thread newThread(Runnable r);
}
JDK线程池:Executors.newSingleThreadExecutor、Executors.newFixedThreadPool等由一个ThreadFactory来创建新的线程,默认情况下为Executors.defaultThreadFactory(),我们可以采用自定义的ThreadFactory工厂,增加对线程创建与销毁等更多的控制(比如上述代码中的内部类CustomThreadFactory即为新建线程的模板)
此处简单提及一下,读者欲了解更多内容可以参考以下文章
(1)http://guojuanjun.blog.51cto.com/277646/650981/
(2)http://ifeve.com/customizing-concurrency-classes-4/
八、扩展ThreadPoolExecutor
九、源码视角
从源码视角分析Executors、ThreadPoolExecutor、ExecuteService、Executor之间的关系,此处简单提及,读者可查看下一节“参考资料”以了解相关内容
(1)Executors
从Java5开始新增了Executors类,它有几个静态工厂方法用来创建线程池,这些静态工厂方法返回一个ExecutorService类型的值,此值即为线程池的引用。
(2)Executor
Executor是一个接口,里面只有一个方法
public interface Executor {
void execute(Runnable command);
}
(3)ExecuteService
ExecuteService也是一个接口,其定义如下:
public interface ExecutorService extends Executor {...}
(4)ThreadPoolExecutor继承AbstractExecutorService,AbstractExecutorService实现ExecutorService接口
public class ThreadPoolExecutor extends AbstractExecutorService {...}
public abstract class AbstractExecutorService implements ExecutorService {...}
十、ExecutorService的生命周期
在本文最开始的那个示例中,有一句代码,如下:
executor.shutdown();
该语句并不是终止线程的运行,而是禁止在这个executor中添加新的任务,下文描述了该语句对于ExecutorService的意义。
十一、参考资料
本文仅简单阐述了Java并发中关于Executors及ThreadPoolExecutor的内容,此处贴出一些优质文章以供读者阅览
(1)http://blog.csdn.net/xiamizy/article/details/40781939
(2)http://www.cnblogs.com/dolphin0520/p/3932921.html
(3)http://www.cnblogs.com/yezhenhan/archive/2012/01/07/2315645.html
(4)http://www.cnblogs.com/guguli/p/5198894.html
java阻塞线程池_线程池解决阻塞方法相关推荐
- java 动态线程池_线程池的参数动态调整
经典面试题 这次的文章还是绕回了我写的第三篇原创文章<有的线程它死了,于是它变成一道面试题>中留下的几个问题: 哎,兜兜转转,走走停停.天道好轮回,苍天饶过谁? 在这篇文章中我主要回答上面 ...
- java开源线程池_线程池 - Java 并发性和多线程 - UDN开源文档
线程池 线程池(Thread Pool)对于限制应用程序中同一时刻运行的线程数很有用.因为每启动一个新线程都会有相应的性能开销,每个线程都需要给栈分配一些内存等等. 我们可以把并发执行的任务传递给一个 ...
- java executors 详解_线程池—Executors 详解
各位志同道合的朋友们大家好,我是一个一直在一线互联网踩坑十余年的编码爱好者,现在将我们的各种经验以及架构实战分享出来,如果大家喜欢,就关注我,一起将技术学深学透,我会每一篇分享结束都会预告下一专题 线 ...
- java executors 详解_线程池Executors详解
为什么要用线程池呢? 一是减少了创建和销毁线程的次数,每个工作线程都可以被重复利用,可执行多个任务; 二是可以根据系统的承受能力,调整线程池中工作线线程的数目,防止因为因为消耗过多的内存,而把服务器累 ...
- postgres 支持的线程数_线程池被打满了怎么处理呢,你是否真的了解线程池?
0.前言 线程池,顾名思义就是线程的池子,在每次需要取线程去执行任务的时候,没必要每次都创建新线程执行,线程池就是起着维护线程的作用,当有任务的时候就取出一个线程执行,如果任务执行完成则把线程放回到池 ...
- threadpoolexecutor创建线程池_线程池ThreadPoolExecutor源码分析
什么是线程池 创建线程要花费昂贵的资源和时间,如果任务来了才创建那么响应时间会变长,而且一个进程能创建的线程数量有限.为了避免这些问题,在程序启动的时候就创建若干线程来响应出来,它们被称为线程池,里面 ...
- mysql 关闭线程池_线程池安全的关闭方式
对于一些定时任务或者网络请求服务将会使用线程池,当应用停机时需要正确安全的关闭线程池,如果处理不当,可能造成数据丢失,业务请求结果不正确等问题. 关闭线程池我们可以选择什么都不做,JVM 关闭时自然的 ...
- springboot 压测 50并发 线程等待_线程池+CountDownLatch——高并发就是这么简单
今天和大家分享的是:在开发服务端API时候,如何合理的运用线程池+CountDownLatch来保证API的高并发访问. 首先,作为Java开发的同学来说,java.util.concurrent并发 ...
- java 线程僵死_线程的生命周期?什么时候会出现僵死进程?
线程的生命周期及五种基本状态 关于Java中线程的生命周期,首先看一下下面这张较为经典的图: 新建(new Thread) 当创建Thread类的一个实例(对象)时,此线程进入新建状态(未被启动).例 ...
最新文章
- 0-1背包使用一维dp数组时为何v要从大到小枚举
- Vb Shell 打开程序 等待运行完毕后再继续
- [vue-element] ElementUI使用表格组件时有遇到过问题吗?
- python学习笔记_序
- mysql 备份的脚本
- 暂无支持此机型的手机版本_miui+app官方版-miui+安卓版 _5577安卓网
- linux 对文件类型的判断
- 深入理解 Laravel 管道
- 异步日志方案log4cpp
- 软件测试 测试用例笔记
- 一二线城市都有哪些互联网公司是大家都想去的?
- RadioButton设置图片大小
- DDR内存的终极优化
- 给学弟学妹肝的一篇Java学习路线,保证学弟学妹们大三大四的时候顺利找到实习 Offer
- HAL库教程6:串口数据接收
- 自有APP上如何运行小游戏?
- web2QQ简单复制
- IO输入输出模型是每个Java开发人员必须理解的重点,深度解析跳槽从开始到结束完整流程
- 破解android锁屏密码
- js怎么判断是否是数组的六种方法