您的位置：首页 > 其它

ThreadPoolExecutor运行原理

2015-10-17 21:02 579 查看

线程池用来管理工作线程，它包含一个队列用来存放等待执行的任务。java的

java.util.concurrent.Executors

提供了实现

java.util.concurrent.Executor

这个接口的方法，用来创建线程池。

下面从一个简单的例子解释它的运行机制。

创建Runnable类

public class WorkerThread implements Runnable {
private String command;
public WorkerThread(String s){
this.command=s;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName()+' Start. Command = '+command);
processCommand();
System.out.println(Thread.currentThread().getName()+' End.');
}
private void processCommand() {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public String toString(){
return this.command;
}
}

给出测试程序，使用Executors框架创建固定大小的线程池。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class SimpleThreadPool {
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
Runnable worker = new WorkerThread('' + i);
executor.execute(worker);
}
executor.shutdown();
while (!executor.isTerminated()) {
}
System.out.println('Finished all threads');
}
}

上面的程序中，创建了固定大小的线程池，能容纳5个工作线程，然后向这个线程池中提交10个任务，因为线程池的大小为5，先执行其中的5个任务，剩余的5个任务处于等待状态，只要其中的一个任务完成结束，处于等待队列的任务的工作线程取出并执行。上面程序的运行结果如下：

pool-1-thread-2 Start. Command = 1

pool-1-thread-4 Start. Command = 3

pool-1-thread-1 Start. Command = 0

pool-1-thread-3 Start. Command = 2

pool-1-thread-5 Start. Command = 4

pool-1-thread-4 End.

pool-1-thread-5 End.

pool-1-thread-1 End.

pool-1-thread-3 End.

pool-1-thread-3 Start. Command = 8

pool-1-thread-2 End.

pool-1-thread-2 Start. Command = 9

pool-1-thread-1 Start. Command = 7

pool-1-thread-5 Start. Command = 6

pool-1-thread-4 Start. Command = 5

pool-1-thread-2 End.

pool-1-thread-4 End.

pool-1-thread-3 End.

pool-1-thread-5 End.

pool-1-thread-1 End.

Finished all threads

Executors

类使用

ThreadPoolExecutor

提供了

ExecutorService

的简单实现1，但是

ThreadPoolExecutor

提供了更多的特征。在代码中，当创建

ThreadPoolExecutor

实例时，可以控制存活线程的数量、限制线程池的大小和实现

RejectedExecutionHandler

处理不能在等待队列中容纳的任务2。

下面先实现

RejectedExecutionHandler

接口的类。

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
public class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor)
{
System.out.println(r.toString() + ' is rejected');
}
}

下面创建监控线程用来打印不同时间段executor的执行情况。

import java.util.concurrent.ThreadPoolExecutor;
public class MyMonitorThread implements Runnable
{
private ThreadPoolExecutor executor;
private int seconds;
private boolean run=true;
public MyMonitorThread(ThreadPoolExecutor executor, int delay)
{
this.executor = executor;
this.seconds=delay;
}
public void shutdown(){
this.run=false;
}
@Override
public void run()
{
while(run){
System.out.println(
String.format('[monitor] [%d/%d] Active: %d, Completed: %d, Task: %d, isShutdown: %s, isTerminated: %s',
this.executor.getPoolSize(),
this.executor.getCorePoolSize(),
this.executor.getActiveCount(),
this.executor.getCompletedTaskCount(),
this.executor.getTaskCount(),
this.executor.isShutdown(),
this.executor.isTerminated()));
try {
Thread.sleep(seconds*1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

下面使用

ThreadPoolExecutor

创建线程池。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class WorkerPool {
public static void main(String args[]) throws InterruptedException{
//RejectedExecutionHandler implementation
RejectedExecutionHandlerImpl rejectionHandler = new RejectedExecutionHandlerImpl();
//Get the ThreadFactory implementation to use
ThreadFactory threadFactory = Executors.defaultThreadFactory();
//creating the ThreadPoolExecutor
ThreadPoolExecutor executorPool = new ThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2), threadFactory, rejectionHandler);
//start the monitoring thread
MyMonitorThread monitor = new MyMonitorThread(executorPool, 3);
Thread monitorThread = new Thread(monitor);
monitorThread.start();
//submit work to the thread pool
for(int i=0; i<10; i++){
executorPool.execute(new WorkerThread('cmd'+i));
}
Thread.sleep(30000);
//shut down the pool
executorPool.shutdown();
//shut down the monitor thread
Thread.sleep(5000);
monitor.shutdown();
}
}

在初始化

ThreadPoolExecutor

时，设置线程池的大小

corePoolSize

为2，这时没有工作线程（work thread）。随着

ThreadPoolExecutor

对任务的提交（submit），新的工作线程被创建，这些线程不会被空闲（idle），直到线程的数量达到

corePoolSize

= 2。接下来新来的任务会被放在工作队列（work queue）中，直到工作队列满了（等待队列大小为2），这时新的工作线程又被创建，线程的数量直到达到

maximumPoolSize

=4，超出的任务有

RejectedExecutionHandlerImpl

处理。

在创建ThreadPoolExecutor时，应该注意上述参数的设置3。

运行结果：

pool-1-thread-1 Start. Command = cmd0

pool-1-thread-4 Start. Command = cmd5

cmd6 is rejected

pool-1-thread-3 Start. Command = cmd4

pool-1-thread-2 Start. Command = cmd1

cmd7 is rejected

cmd8 is rejected

cmd9 is rejected

[monitor] [0/2] Active: 4, Completed: 0, Task: 6, isShutdown: false, isTerminated: false

[monitor] [4/2] Active: 4, Completed: 0, Task: 6, isShutdown: false, >isTerminated: false

pool-1-thread-4 End.

pool-1-thread-1 End.

pool-1-thread-2 End.

pool-1-thread-3 End.

pool-1-thread-1 Start. Command = cmd3

pool-1-thread-4 Start. Command = cmd2

[monitor] [4/2] Active: 2, Completed: 4, Task: 6, isShutdown: false, isTerminated: false

[monitor] [4/2] Active: 2, Completed: 4, Task: 6, isShutdown: false, isTerminated: false

pool-1-thread-1 End.

pool-1-thread-4 End.

[monitor] [4/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false

[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false

[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false

[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, >isTerminated: false

[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false

[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false

[monitor] [0/2] Active: 0, Completed: 6, Task: 6, isShutdown: true, isTerminated: true

[monitor] [0/2] Active: 0, Completed: 6, Task: 6, isShutdown: true, isTerminated: true

ThreadPoolExecutor

构造方法

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long
keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory, RejectedExecutionHandler handler)


参数	含义
corePoolSize	线程池的基本大小，如果队列中任务已满，并且当前线程个数小于 maximumPoolSize ，那么会创建新的线程来执行任务
maximumPoolSize	线程池中允许的最大线程数，当线程数量大于 maximumPoolSize ，超出的线程由丢弃机制来处理
workQueue	等待队列，当线程的数量达到 corePoolSize 时，就向该等待队列放入线程信息（默认为一个 LinkedBlockingQueue ）
keepAliveTime	当线程池中的线程数量大于 corePoolSize 的时候，多余的线程会等待 keepAliveTime 长的时间，如果无请求可处理就自行销毁
threadFactory 4	用来构造 Thread 对象，主要包括线程名称，守护线程的状态，线程的优先级以及设置 UncaughtExceptionHandler
handler	超过 maximumPoolSize 之后丢弃处理的方法，java提供了5种丢弃处理的方法，也可以自定义，实现 RejectedExecutionHandler 接口

参考资料：

Java Thread Pool Example using Executors and ThreadPoolExecutor

A.

Executors.newCachedThreadPool()

无限大小的线程池，线程会自动重用。相当于

new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>())

，

SynchronousQueue

可以存储元素，但是总是报“I am full”，导致

ThreadPoolExecutor

创建新的线程去

SynchronousQueue

中拉取任务并运行该任务。线程数量是无界的，队列是有界的，用它处理大批量任务时比较危险的。

B.

Executors.newFixedThreadPool(int)

固定线程数的线程池。相当于

new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())

，线程数量是有界的，而工作队列是无界的，如果负载超过了当前的处理能力，那么工作队列会堆积越来越多的任务，有可能会导致内存不足的问题。

C.

Executors.newSingleThreadExecutor()

单线程执行器。 ↩
如果线程池中线程数目已达到

maximumPoolSize

，则提交的任务交由

RejectedExecutionHandler

处理丢弃。对于任务丢弃，

ThreadPoolExecutor

以内部类的形式实现了4个策略：

-

CallerRunsPolicy

：提交任务的线程自己负责执行这个任务；

-

AbortPolicy

：使

Executor

抛出异常，通过异常做处理；

-

DiscardPolicy

：丢弃提交的任务；

-

DiscardOldestPolicy

：丢弃掉队列中最早加入的任务。

在调用构造方法时，参数中未指定

RejectedExecutionHandler

情况下，默认采用

AbortPolicy

。 ↩
创建

ThreadPoolExecutor

应该避免设置

corePoolSize

过小，

maximumPoolSize

过大以及过大的无界队列

LinkedBlockingQueue

，容易造成工作队列未满，而抛出

OutOfMemoryError

异常。 ↩
给线程设置一个好的有状态的名字，如

Thread.currentThread().setName(Context + TID + Params + current Time,..)

，具体请参考这里。当线程中运行一个任务时发生未捕获异常，默认情况下线程终止，异常信息打印在终端，这种默认的行为并不恰当。在创建线程时可以定制未捕获异常的处理。如：

new UncaughtExceptionHandler() {

@Override    public void uncaughtException(Thread t, Throwable e) {

System.err.println(String.format( "Thread %s threw exception - %s", t.getName(), e.getMessage())) } }

↩

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签：

相关文章推荐

新的分享

章节导航