Java高并发编程：线程池

1. Executor类的继承结构

2. ThreadPoolExecutor

2.1 构造方法

public ThreadPoolExecutor(
int corePoolSize, //核心线程数
int maximumPoolSize, //最大线程数
long keepAliveTime, //保持时间
TimeUnit unit, //时间单位
BlockingQueue<Runnable> workQueue, //阻塞队列
ThreadFactory threadFactory, //线程工厂
RejectedExecutionHandler handler //异常捕获器
)

2.1.1 int corePoolSize

2.1.2 int maximumPoolSize

2.1.3 long keepAliveTime

2.1.4 TimeUnit unit

2.1.5 RejectedExecutionHandler

3. 任务提交给线程池之后的处理策略

4. 阻塞队列的介绍

4.1 BlockingQueue

4.2 排队策略

直接提交

无界队列

有界队列

4.3 BlockingQueue

BlockingQueue

null

add

put

offer

null

NullPointerException

null

poll

BlockingQueue

remainingCapacity

put

BlockingQueue

Integer.MAX_VALUE

BlockingQueue

Collection

remove(x)

BlockingQueue

addAll

containsAll

retainAll

removeAll

c

addAll(c)

Blo
1370b
ckingQueue

4.4 BlockingDeque

4.5 ArrayBlockingQueue

public class BlockingQueueCondition {
public static void main(String[] args) {
ExecutorService service = Executors.newSingleThreadExecutor();
final Business3 business = new Business3();
service.execute(new Runnable(){
public void run() {
for(int i=0;i<50;i++){
business.sub();
}
}
});
for(int i=0;i<50;i++){
business.main();
}
}
}
class Business3{
BlockingQueue subQueue  = new ArrayBlockingQueue(1);
BlockingQueue mainQueue = new ArrayBlockingQueue(1);
{
try {
mainQueue.put(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void sub(){
try
{
mainQueue.take();
for(int i=0;i<10;i++){
System.out.println(Thread.currentThread().getName() + " : " + i);
}
subQueue.put(1);
}catch(Exception e){
}
}
public void main(){
try
{
subQueue.take();
for(int i=0;i<5;i++){
System.out.println(Thread.currentThread().getName() + " : " + i);
}
mainQueue.put(1);
}catch(Exception e){
}
}
}

pool-1-thread-1 : 0
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 : 5
pool-1-thread-1 : 6
pool-1-thread-1 : 7
pool-1-thread-1 : 8
pool-1-thread-1 : 9
main : 0
main : 1
main : 2
main : 3
main : 4
pool-1-thread-1 : 0
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 : 5
pool-1-thread-1 : 6
pool-1-thread-1 : 7
pool-1-thread-1 : 8
pool-1-thread-1 : 9
main : 0
main : 1
main : 2
main : 3
main : 4
pool-1-thread-1 : 0
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 : 5
pool-1-thread-1 : 6
pool-1-thread-1 : 7
pool-1-thread-1 : 8
pool-1-thread-1 : 9
...

4.6 LinkedBlockingQueue

4.7 SynchronousQueue

4.8 DelayQueue

4.9 PriorityBlockingQueue

4.10 生产者消费者模式

5. 线程池工具类Executors

6. 线程池

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ThreadPoolTest {
public static void main(String[] args){

//创建固定大小的线程池，这里只能完成3个任务
//ExecutorService threadPool = Executors.newFixedThreadPool(3);

//创建缓存线程池，根据任务来自动创建线程的数量，可以完成创建的所有任务
//ExecutorService threadPool = Executors.newCachedThreadPool();

//创建单一线程池（始终保持线程池中有一个线程存活。当唯一线程死去，会创建新的继任者、
ExecutorService threadPool = Executors.newSingleThreadExecutor();

for(int i=1;i<=10;i++){
　　//内部类不能访问外部类的局部变量，所以i要定义为final，又由于i++.
　　//所以在循环内部定义一个变量接收i
final int task = i;
threadPool.execute(new Runnable() {

@Override
public void run() {
for(int j=1;j<=10;j++){
System.out.println(Thread.currentThread().getName()
+" is looping of "+ j+"  for task of " +task);
}

}
});
}
//验证10个任务都提交给了线程池
System.out.println("all of 10 tasks have committed! ");
//threadPool.shutdown();        //等任务完成后，杀死线程、
//threadPool.shutdownNow();     //立即停止线程

//用线程池启动定时器

Executors.newScheduledThreadPool(3).schedule(
new Runnable() {  //任务
@Override
public void run() {
System.out.println("bombing!");
}
},
5,  //5秒以后执行
TimeUnit.SECONDS);  //单位

//在某个时间执行一次后，再指定后续的执行间隔时间
Executors.newScheduledThreadPool(3).scheduleAtFixedRate(new Runnable(){

@Override
public void run() {
System.out.println("bombing!");
}

}, 10,   //第一次在10秒时爆炸
3,          //以后每隔3秒爆炸一次。
TimeUnit.SECONDS);

}
}

7. 线程池的简单使用

/**
* 一个简易的线程池管理类，提供三个线程池
*/
public class ThreadManager {
public static final String DEFAULT_SINGLE_POOL_NAME = "DEFAULT_SINGLE_POOL_NAME";

private static ThreadPoolProxy mLongPool = null;
private static Object mLongLock = new Object();

private static ThreadPoolProxy mShortPool = null;
private static Object mShortLock = new Object();

private static ThreadPoolProxy mDownloadPool = null;
private static Object mDownloadLock = new Object();

private static Map<String, ThreadPoolProxy> mMap = new HashMap<String, ThreadPoolProxy>();
private static Object mSingleLock = new Object();

/** 获取下载线程 */
public static ThreadPoolProxy getDownloadPool() {
synchronized (mDownloadLock) {
if (mDownloadPool == null) {
mDownloadPool = new ThreadPoolProxy(3, 3, 5L);
}
return mDownloadPool;
}
}

/** 获取一个用于执行长耗时任务的线程池，避免和短耗时任务处在同一个队列而阻塞了重要的短耗时任务，通常用来联网操作 */
public static ThreadPoolProxy getLongPool() {
synchronized (mLongLock) {
if (mLongPool == null) {
mLongPool = new ThreadPoolProxy(5, 5, 5L);
}
return mLongPool;
}
}

/** 获取一个用于执行短耗时任务的线程池，避免因为和耗时长的任务处在同一个队列而长时间得不到执行，通常用来执行本地的IO/SQL */
public static ThreadPoolProxy getShortPool() {
synchronized (mShortLock) {
if (mShortPool == null) {
mShortPool = new ThreadPoolProxy(2, 2, 5L);
}
return mShortPool;
}
}

/** 获取一个单线程池，所有任务将会被按照加入的顺序执行，免除了同步开销的问题 */
public static ThreadPoolProxy getSinglePool() {
return getSinglePool(DEFAULT_SINGLE_POOL_NAME);
}

/** 获取一个单线程池，所有任务将会被按照加入的顺序执行，免除了同步开销的问题 */
public static ThreadPoolProxy getSinglePool(String name) {
synchronized (mSingleLock) {
ThreadPoolProxy singlePool = mMap.get(name);
if (singlePool == null) {
singlePool = new ThreadPoolProxy(1, 1, 5L);
mMap.put(name, singlePool);
}
return singlePool;
}
}

public static class ThreadPoolProxy {
private ThreadPoolExecutor mPool;
private int mCorePoolSize;
private int mMaximumPoolSize;
private long mKeepAliveTime;

private ThreadPoolProxy(int corePoolSize, int maximumPoolSize, long keepAliveTime) {
mCorePoolSize = corePoolSize;
mMaximumPoolSize = maximumPoolSize;
mKeepAliveTime = keepAliveTime;
}

/** 执行任务，当线程池处于关闭，将会重新创建新的线程池 */
public synchronized void execute(Runnable run) {
if (run == null) {
return;
}
if (mPool == null || mPool.isShutdown()) {
//参数说明
//当线程池中的线程小于mCorePoolSize，直接创建新的线程加入线程池执行任务
//当线程池中的线程数目等于mCorePoolSize，将会把任务放入任务队列BlockingQueue中
//当BlockingQueue中的任务放满了，将会创建新的线程去执行，
//但是当总线程数大于mMaximumPoolSize时，将会抛出异常，交给RejectedExecutionHandler处理
//mKeepAliveTime是线程执行完任务后，且队列中没有可以执行的任务，存活的时间，后面的参数是时间单位
//ThreadFactory是每次创建新的线程工厂
mPool = new ThreadPoolExecutor(mCorePoolSize, mMaximumPoolSize, mKeepAliveTime, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), Executors.defaultThreadFactory(), new AbortPolicy());
}
mPool.execute(run);
}

/** 取消线程池中某个还未执行的任务 */
public synchronized void cancel(Runnable run) {
if (mPool != null && (!mPool.isShutdown() || mPool.isTerminating())) {
mPool.getQueue().remove(run);
}
}

/** 取消线程池中某个还未执行的任务 */
public synchronized boolean contains(Runnable run) {
if (mPool != null && (!mPool.isShutdown() || mPool.isTerminating())) {
return mPool.getQueue().contains(run);
} else {
return false;
}
}

/** 立刻关闭线程池，并且正在执行的任务也将会被中断 */
public void stop() {
if (mPool != null && (!mPool.isShutdown() || mPool.isTerminating())) {
mPool.shutdownNow();
}
}

/** 平缓关闭单任务线程池，但是会确保所有已经加入的任务都将会被执行完毕才关闭 */
public synchronized void shutdown() {
if (mPool != null && (!mPool.isShutdown() || mPool.isTerminating())) {
mPool.shutdownNow();
}
}
}
}