线程池、饱和策略详解

一、序言

当我们需要使用线程的时候,我们可以新建一个线程,然后显式调用线程的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提供不同的线程池机制。

image

四、简述线程池的属性

线程池的属性

五、详解ThreadPoolExecutor##

上文提到可以通过显式的ThreadPoolExecutor构造函数来构造特定形式的线程池,ThreadPoolExecutorjava.util.concurrent包以内部线程池的形式对外提供线程池管理、线程调度等服务,此处我们来了解一下ThreadPoolExecutor

  • (1)一般使用方式:
ExecutorService exec = new ThreadPoolExecutor(8, 8, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(100), new ThreadPoolExecutor.CallerRunsPolicy());

下文详解此示例涉及的一些内容

  • 2)构造函数的声明:
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,  
                              TimeUnit unit,  
                              BlockingQueue<Runnable> 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)任务阻塞队列选择机制

image
  • (5)简述SynchronousQueue
image

注:此处贴出SynchronousQueue的使用示例,示例中使用了Semaphore,更多关于SynchronousQueueSemaphore的内容请参考其他文章

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<String> sq = new SynchronousQueue<String>(); // 定义一个数量为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)四种饱和策略

image
image
image

*(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)通过修改参数的方式达到定制目的

image
image

*(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<Runnable>(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<Runnable>(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,前文提到该参数是新建线程的工厂,此处进一步解说该参数。

ThreadFactoryjava.util.concurrent包下创建线程工厂的接口,ThreadFactory接口源码如下:

public interface ThreadFactory {  
    Thread newThread(Runnable r);  
}

JDK线程池:Executors.newSingleThreadExecutor、``Executors.newFixedThreadPool等由一个ThreadFactory来创建新的线程,默认情况下为Executors.defaultThreadFactory(),我们可以采用自定义的ThreadFactory工厂,增加对线程创建与销毁等更多的控制(比如上述代码中的内部类CustomThreadFactory即为新建线程的模板)

此处简单提及一下,读者欲了解更多内容可以参考以下文章

八、扩展ThreadPoolExecutor##

image

九、源码视角##

从源码视角分析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的意义。

image

十一、参考资料##

本文仅简单阐述了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

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 216,402评论 6 499
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 92,377评论 3 392
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 162,483评论 0 353
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 58,165评论 1 292
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 67,176评论 6 388
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 51,146评论 1 297
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 40,032评论 3 417
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 38,896评论 0 274
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 45,311评论 1 310
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,536评论 2 332
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,696评论 1 348
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 35,413评论 5 343
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 41,008评论 3 325
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,659评论 0 22
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,815评论 1 269
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 47,698评论 2 368
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 44,592评论 2 353

推荐阅读更多精彩内容

  • 第一部分 来看一下线程池的框架图,如下: 1、Executor任务提交接口与Executors工具类 Execut...
    压抑的内心阅读 4,261评论 1 24
  • 前段时间遇到这样一个问题,有人问微信朋友圈的上传图片的功能怎么做才能让用户的等待时间较短,比如说一下上传9张图片,...
    加油码农阅读 1,201评论 0 2
  • 一.Java中的ThreadPoolExecutor类 java.uitl.concurrent.ThreadPo...
    谁在烽烟彼岸阅读 645评论 0 0
  • 你对于所有的抱怨都来自于你的无知 清明小长假回来后,无比热爱周口这个城市。走出出站口,下意识的拥抱城市,这时候,旅...
    0852滴女咿呀阅读 126评论 0 0
  • 属于我们的过去过去了 你像患了失忆症 只我一个人回忆 曾以为永生难忘的事啊 数着数着就丢掉了 丢掉了日,丢掉了月,...
    二三二六阅读 226评论 0 0