ThreadPoolExecutor详解

本文系转载，原文地址：http://825635381.iteye.com/blog/2184680

ThreadPoolExecutor机制

一、概述

1、ThreadPoolExecutor作为java.util.concurrent包对外提供基础实现，以内部线程池的形式对外提供管理任务执行，线程调度，线程池管理等等服务

2、Executors方法提供的线程服务，都是通过参数设置来实现不同的线程池机制

3、先来了解其线程池管理的机制，有助于正确使用，避免错误使用导致严重故障。同时可以根据自己的需求实现自己的线程池

二、核心构造方法讲解

1.下面是ThreadPoolExecutor最核心的构造方法:

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;

    }

2.构造方法参数讲解

    corePoolSize--核心线程池大小

    maximumPoolSize--最大线程池大小

    keepAliveTime--线程池中超过corePoolSize数目的空闲线程最大存活时间，

                          可以allowCoreThreadTimeOut(true)使得核心   线程有效时间

    TimeUnitkeepAliveTime--时间单位

    workQueue--阻塞任务队列

    threadFactory--新建线程工厂

    RejectedExecutionHandler--当提交任务数超过maxmumPoolSize+workQueue之和时，

                                             任务会交给RejectedExecutionHandler来处理

3.线程池规则：

    线程池的线程执行规则跟任务队列有很大的关系

1)假设任务队列没有大小限制：

    如果线程数量<=核心线程数量，那么直接启动一个核心线程来执行任务，不会放入队列中。

    如果线程数量>核心线程数，但<=最大线程数，并且任务队列是LinkedBlockingDeque的时候，超过核心线程数量的任务会放在任务队列中排队。

    如果线程数量>核心线程数，但<=最大线程数，并且任务队列是SynchronousQueue的时候，线程池会创建新线程执行任务，这些任务也不会被放在任务队列中。这些线程属于非核心线程，在任务完成后，闲置时间达到了超时时间就会被清除。

    如果线程数量>核心线程数，并且>最大线程数，当任务队列是LinkedBlockingDeque，会将超过核心线程的任务放在任务队列中排队。也就是当任务队列是LinkedBlockingDeque并且没有大小限制时，线程池的最大线程数设置是无效的，他的线程数最多不会超过核心线程数。

    如果线程数量>核心线程数，并且>最大线程数，当任务队列是SynchronousQueue的时候，会因为线程池拒绝添加任务而抛出异常。

2)任务队列大小有限时

    当LinkedBlockingDeque塞满时，新增的任务会直接创建新线程来执行，当创建的线程数量超过最大线程数量时会抛异常。

    SynchronousQueue没有数量限制。因为他根本不保持这些任务，而是直接交给线程池去执行。当任务数量超过最大线程数时，新提交任务由RejectedExecutionHandler处理，会直接抛异常。

3)当线程池中超过corePoolSize线程，空闲时间达到keepAliveTime时，关闭空闲线程

4)当设置allowCoreThreadTimeOut(true)时，线程池中corePoolSize线程空闲时间达到keepAliveTime也将关闭

4.线程管理机制图示：

三、Executors提供的线程池配置方案

1、构造一个固定线程数目的线程池，配置的corePoolSize与maximumPoolSize大小相同，同时使用了一个无界LinkedBlockingQueue存放阻塞任务，因此多余的任务将存在再阻塞队列，不会由RejectedExecutionHandler处理

public static ExecutorService newFixedThreadPool(int nThreads) {

    return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());

}

2、构造一个缓冲功能的线程池，配置corePoolSize=0，maximumPoolSize=Integer.MAX_VALUE，keepAliveTime=60s,以及一个无容量的阻塞队列 SynchronousQueue，因此任务提交之后，将会创建新的线程执行；线程空闲超过60s将会销毁

public static ExecutorService newCachedThreadPool() {

    return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue());

}

3、构造一个只支持一个线程的线程池，配置corePoolSize=maximumPoolSize=1，无界阻塞队列LinkedBlockingQueue；保证任务由一个线程串行执行

public static ExecutorService newSingleThreadExecutor() {

    return new FinalizableDelegatedExecutorService(

        new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()));

}

4、构造有定时功能的线程池，配置corePoolSize，无界延迟阻塞队列DelayedWorkQueue；有意思的是：maximumPoolSize=Integer.MAX_VALUE，由于DelayedWorkQueue是无界队列，所以这个值是没有意义的

public static ScheduledExecutorServicenewScheduledThreadPool(int corePoolSize) {

    return new ScheduledThreadPoolExecutor(corePoolSize);

}

public static ScheduledExecutorServicenewScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory) {

    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);

}

public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) {

    super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,

        new DelayedWorkQueue(), threadFactory);

}

四、定制属于自己的非阻塞线程池

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;

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

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{

    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() {

    public void run() {

        try {

            Thread.sleep(3000);

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        System.out.println("running=====");

}

});

}

// 2.销毁----此处不能销毁,因为任务没有提交执行完,如果销毁线程池,任务也就无法执行了

// exec.destory();

try {

    Thread.sleep(10000);

} catch (InterruptedException e) {

    e.printStackTrace();

}

}

}

方法中建立一个核心线程数为30个，缓冲队列有10个的线程池。每个线程任务，执行时会先睡眠3秒，保证提交10任务时，线程数目被占用完，再提交30任务时，阻塞队列被占用完，，这样提交第41个任务是，会交给CustomRejectedExecutionHandler 异常处理类来处理。

提交任务的代码如下：

public void execute(Runnable command) {

if (command == null)

throw new NullPointerException();

int c = ctl.get();

if (workerCountOf(c) < corePoolSize) {

if (addWorker(command, true))

return;

c = ctl.get();

}

if (isRunning(c) && workQueue.offer(command)) {

int recheck = ctl.get();

if (! isRunning(recheck) && remove(command))

    reject(command);

else if (workerCountOf(recheck) == 0)

    addWorker(null, false);

}

else if (!addWorker(command, false))

    reject(command);

}

注意：41以后提交的任务就不能正常处理了，因为，execute中提交到任务队列是用的offer方法，如上面代码，这个方法是非阻塞的，所以就会交给CustomRejectedExecutionHandler 来处理，所以对于大数据量的任务来说，这种线程池，如果不设置队列长度会OOM，设置队列长度，会有任务得不到处理，接下来我们构建一个阻塞的自定义线程池

五、定制属于自己的阻塞线程池

package com.tongbanjie.trade.test.commons;

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;

public void init() {

pool = new ThreadPoolExecutor(

1,

3,

30,

TimeUnit.MINUTES,

new ArrayBlockingQueue(5),

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) {

try {

// 核心改造点，由blockingqueue的offer改成put阻塞方法

executor.getQueue().put(r);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

// 测试构造的线程池

public static void main(String[] args) {

CustomThreadPoolExecutorexec = 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 {

System.out.println(">>>task is running=====");

TimeUnit.SECONDS.sleep(10);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

});

}

// 2.销毁----此处不能销毁,因为任务没有提交执行完,如果销毁线程池,任务也就无法执行了

// exec.destory();

try {

Thread.sleep(10000);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

解释：当提交任务被拒绝时，进入拒绝机制，我们实现拒绝方法，把任务重新用阻塞提交方法put提交，实现阻塞提交任务功能，防止队列过大，OOM，提交被拒绝方法在下面

public void execute(Runnable command) {

if (command == null)

    throw new NullPointerException();

int c = ctl.get();

if (workerCountOf(c) < corePoolSize) {

    if (addWorker(command, true))

    return;

c = ctl.get();

}

if (isRunning(c) && workQueue.offer(command)) {

    int recheck = ctl.get();

    if (! isRunning(recheck) && remove(command))

        reject(command);

    else if (workerCountOf(recheck) == 0)

        addWorker(null, false);

} else if (!addWorker(command, false))

    // 进入拒绝机制， 我们把runnable任务拿出来，重新用阻塞操作put，来实现提交阻塞功能

    reject(command);

}

总结：

1、用ThreadPoolExecutor自定义线程池，看线程是的用途，如果任务量不大，可以用无界队列，如果任务量非常大，要用有界队列，防止OOM

2、如果任务量很大，还要求每个任务都处理成功，要对提交的任务进行阻塞提交，重写拒绝机制，改为阻塞提交。保证不抛弃任何任务

3、最大线程数一般设为2N+1最好，N是CPU核数

4、核心线程数，看应用，如果是任务，一天跑一次，设置为0，合适，因为跑完就停掉了，如果是常用线程池，看任务量，是保留一个核心还是几个核心线程数

5、如果要获取任务执行结果，用CompletionService，但是注意，获取任务的结果的要重新开一个线程获取，如果在主线程获取，就要等任务都提交后才获取，就会阻塞大量任务结果，队列过大OOM，所以最好异步开个线程获取结果

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 213,711评论 6赞 493
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 91,079评论 3赞 387
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 159,194评论 0赞 349
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 57,089评论 1赞 286
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 66,197评论 6赞 385
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,306评论 1赞 292
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,338评论 3赞 412
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,119评论 0赞 269
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,541评论 1赞 306
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,846评论 2赞 328
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,014评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,694评论 4赞 337
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,322评论 3赞 318
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,026评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,257评论 1赞 267
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,863评论 2赞 365
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,895评论 2赞 351

推荐阅读更多精彩内容