正文

Java 并发编程——Executor框架和线程池原理

IP云
IP云V管理员/5 阅读
07/25
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

Java 并发编程系列文章

Java 并发基础——线程安全性

Java 并发编程——Callable+Future+FutureTask

java 并发编程——Thread 源码重新学习

java并发编程——通过ReentrantLock,Condition实现银行存取款

Java并发编程——BlockingQueue

Java 并发编程——Executor框架和线程池原理

Eexecutor作为灵活且强大的异步执行框架，其支持多种不同类型的任务执行策略，提供了一种标准的方法将任务的提交过程和执行过程解耦开发，基于生产者-消费者模式，其提交任务的线程相当于生产者，执行任务的线程相当于消费者，并用Runnable来表示任务，Executor的实现还提供了对生命周期的支持，以及统计信息收集，应用程序管理机制和性能监视等机制。

下面这段代码中将多个任务放到了线程池中执行：

static class MyRunnable implements Runnable{
@Override
public void run() {
for(int i=0;i<100;i++)
System.out.println(Thread.currentThread().getName()+": "+i);
}
}static class MyThread extends Thread {
public MyThread(String in){
super(in);
}
@Override
public void run(){
System.out.println(Thread.currentThread().getName());
}
}public static void main(String[] args) {
ExecutorService pool3 = Executors.newFixedThreadPool(2);
pool3.execute(new MyThread("t1"));
pool3.execute(new MyThread("t2"));
pool3.execute(new MyThread("t3"));
pool3.execute(new MyThread("t4"));
pool3.shutdown(); ExecutorService pool2 = Executors.newFixedThreadPool(2);
pool2.submit(new MyRunnable());
pool2.submit(new MyRunnable());
pool2.shutdownNow();
}

Executor：一个接口，其定义了一个接收Runnable对象的方法executor，其方法签名为executor(Runnable command),

ExecutorService：是一个比Executor使用更广泛的子类接口，其提供了生命周期管理的方法，以及可跟踪一个或多个异步任务执行状况返回Future的方法

AbstractExecutorService：ExecutorService执行方法的默认实现

ScheduledExecutorService：一个可定时调度任务的接口

ScheduledThreadPoolExecutor：ScheduledExecutorService的实现，一个可定时调度任务的线程池

ThreadPoolExecutor：线程池，可以通过调用Executors以下静态工厂方法来创建线程池并返回一个ExecutorService对象：

1. Executor 接口

Executor是java.util.concurrent 包下的一个接口。

public interface Executor { /**
* Executes the given command at some time in the future. The command
* may execute in a new thread, in a pooled thread, or in the calling
* thread, at the discretion of the {@code Executor} implementation.
*
* @param command the runnable task
* @throws RejectedExecutionException if this task cannot be
* accepted for execution
* @throws NullPointerException if command is null
*/
void execute(Runnable command);
}

该接口十分简单，只有一个执行的方法 execute() 方法。

2. ExecutorService接口

为了充分理解ExecutorService接口建议先了解：Java 并发编程——Callable+Future+FutureTask

public interface ExecutorService extends Executor；

void shutdown();

启动一次顺序关闭，执行以前提交的任务，但不接受新任务。

List<Runnable>  shutdownNow()

试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表。

boolean isShutdown()

*
* @return {@code true} if this executor has been shut down
*/

boolean isTerminated()

@return {@code true} if all tasks have completed following shut down

boolean isTerminated()

只有当shutdown()或者shutdownnow()被调用，而且所有任务都执行完成后才会返回true。

boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;

Blocks until all tasks have completed execution after a shutdown
* request, or the timeout occurs, or the current thread is
* interrupted, whichever happens first.

<T> Future<T> submit(Callable<T> task)<T> Future<T> submit(Runnable task, T result)Future<?Future> submit(Runnable task)

都是提交一个任务等待执行，只不过第二个函数Future.get()返回值为result。第三个函数Future.get()返回值为null

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;

相当于submit()函数，不同之处是它可以同时提交多个任务，并将Future列表返回，使用者可以遍历List中的元素进行.get操作。

<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;

<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;

同时提交多个任务，返回第一个执行完成的结果。

ExecutorService中execute()函数和submit()函数的区别（开头线程池使用的代码中使用了这两种方法执行一个任务）

1. 接收的参数不一样，execute接口只能接收Runnable向，而submit接口可以接收多种类型的对象

2.返回值不同，execute没有返回值，而submit返回一个Future对象

3.Exception处理

There is a difference when looking at exception handling. If your tasks throws an exception and if it was submitted with execute this exception will go to the uncaught exception handler (when you don't have provided one explicitly, the default one will just print the stack trace to System.err). If you submitted the task with submit any thrown exception, checked or not, is then part of the task's return status. For a task that was submitted with submit and that terminates with an exception, the Future.get will rethrow this exception, wrapped in an ExecutionException.

3. AbstractExecutorService

通过这个类的名字可能就大概知道了这个类的作用，它实现了ExecutorService中的部分方法，对于继承它的类可以减少实现的代码。该抽象类中并没有存放任务或者线程的数组或者Collection所以，对线程队列的具体管理AbstractExecutorService类并不涉及。

下面看一下里面几个关键接口的实现：

当向线程池中提交一个任务时，它实际上还是调用execute接口去执行的，然后将执行的结果（如果有的话）一个Future对象返回给任务提交方。

public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);

execute(ftask);

return ftask;
}

execute()的实现并没有放在当前的抽象类中实现，而是让子类去实现。

再看一下invokeAll的执行过程：

public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
ArrayList<Future<T>> futures = new ArrayList<>(tasks.size());
try {

// 将tasks转换成Futrues类型

for (Callable<T> t : tasks) {
RunnableFuture<T> f = newTaskFor(t);
futures.add(f);
execute(f);
}

 

// 执行future中的get函数

for (int i = 0, size = futures.size(); i < size; i++) {
Future<T> f = futures.get(i);
if (!f.isDone()) {
try { f.get(); }
catch (CancellationException ignore) {}
catch (ExecutionException ignore) {}
}
}
return futures;
} catch (Throwable t) {
cancelAll(futures);
throw t;
}
}

其它函数的实现原理基本相同，参考源码。

4. ThreadPoolExecutor

public class ThreadPoolExecutor extends AbstractExecutorService

为什么需要线程池

使用线程的时候就去创建一个线程，这样实现起来非常简便，但是就会有一个问题：如果并发的线程数量很多，并且每个线程都是执行一个时间很短的任务就结束了，这样频繁创建线程就会大大降低系统的效率，因为频繁创建线程和销毁线程需要时间。

构造方法

ThreadPoolExecutor是线程池的真正实现，他通过构造方法的一系列参数，来构成不同配置的线程池。常用的构造方法有下面四个：

ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue)

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

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

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

构造方法参数说明

  • corePoolSize

    核心线程数，默认情况下核心线程会一直存活，即使处于闲置状态也不会受存keepAliveTime限制。除非将allowCoreThreadTimeOut设置为true。

  • maximumPoolSize

    线程池所能容纳的最大线程数。超过这个数的线程将被阻塞。当任务队列为没有设置大小的LinkedBlockingDeque时，这个值无效。

  • keepAliveTime

    非核心线程的闲置超时时间，超过这个时间就会被回收。（当线程数没有超过核心线程数时，这个时间没有任何意义）

  • unit

    指定keepAliveTime的单位，如TimeUnit.SECONDS。当将allowCoreThreadTimeOut设置为true时对corePoolSize生效。

  • workQueue

    线程池中的任务队列.

    常用的有三种队列，SynchronousQueue,LinkedBlockingDeque,ArrayBlockingQueue。

  • threadFactory

    线程工厂，提供创建新线程的功能。ThreadFactory是一个接口，只有一个方法。

public interface RejectedExecutionHandler {
void rejectedExecution(Runnable var1, ThreadPoolExecutor var2);
}

当线程池中的资源已经全部使用，添加新线程被拒绝时，会调用RejectedExecutionHandler的rejectedExecution方法。

下图为线程池的主要结构：

一定要注意一个概念，即存在于线程池中容器的一定是Thread对象，而不是你要求运行的任务（所以叫线程池而不叫任务池也不叫对象池）；你要求运行的任务将被线程池分配给某一个空闲的Thread运行。

下面这例子很好的说明了ThreadpoolExecutor的用法：

public class TestThreadPoolExecutor { static class MyTask implements Runnable {
private int taskNum; public MyTask(int num) {
this.taskNum = num;
} @Override
public void run() {
System.out.println("正在执行task "+taskNum);
try {
Thread.currentThread().sleep(4000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("task "+taskNum+"执行完毕");
}
} public static void main(String[] args) {
// 核心线程数为 5
// 最大线程数为 10(最多同时运行10个线程)
// 非核心线程没有任务执行时，最多等待200ms
// 等待队列
ThreadPoolExecutor executor = new ThreadPoolExecutor(5,10,200,
TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(5));
for(int i=0;i<15;i++){
executor.submit(new MyTask(i));
System.out.println("线程池中线程数目："+executor.getPoolSize()+"，队列中等待执行的任务数目："+
executor.getQueue().size()+"，已执行玩别的任务数目："+executor.getCompletedTaskCount());
}
}
}

执行结果如下：

public class TestThreadPoolExecutor { static class MyTask implements Runnable {
private int taskNum; public MyTask(int num) {
this.taskNum = num;
} @Override
public void run() {
System.out.println("正在执行task "+taskNum);
try {
Thread.currentThread().sleep(4000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("task "+taskNum+"执行完毕");
}
} public static void main(String[] args) {
// 核心线程数为 5
// 最大线程数为 10(最多同时运行10个线程)
// 非核心线程没有任务执行时，最多等待200ms
// 等待队列
ThreadPoolExecutor executor = new ThreadPoolExecutor(5,10,200,
TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(5));
for(int i=0;i<15;i++){
executor.submit(new MyTask(i));
System.out.println("线程池中线程数目："+executor.getPoolSize()+"，队列中等待执行的任务数目："+
executor.getQueue().size()+"，已执行玩别的任务数目："+executor.getCompletedTaskCount());
}
}
}

这个例子很好的说明了线程池的执行逻辑：当核心线程池为满时，开启线程执行提交的任务，当核心线程池满时，将任务放到等待队列中等待执行。当队列已满时，开启线程去执行任务（没有达到最大线程数10），当开启线程执行完成后执行等待队列中的线程。

线程池中 核心线程、线程队列、最大线程的运行准则：

1、首先可以通过线程池提供的submit()方法或者execute()方法，要求线程池执行某个任务。线程池收到这个要求执行的任务后，会有几种处理情况：
1.1、如果当前线程池中运行的线程数量还没有达到corePoolSize大小时，线程池会创建一个新的线程运行你的任务，无论之前已经创建的线程是否处于空闲状态。
1.2、如果当前线程池中运行的线程数量已经达到设置的corePoolSize大小，线程池会把你的这个任务加入到等待队列中。直到某一个的线程空闲了，线程池会根据设置的等待队列规则，从队列中取出一个新的任务执行。
1.3、如果根据队列规则，这个任务无法加入等待队列。这时线程池就会创建一个“非核心线程”直接运行这个任务。注意，如果这种情况下任务执行成功，那么当前线程池中的线程数量一定大于corePoolSize。
1.4、如果这个任务，无法被“核心线程”直接执行，又无法加入等待队列，又无法创建“非核心线程”直接执行，且你没有为线程池设置RejectedExecutionHandler。这时线程池会抛出RejectedExecutionException异常，即线程池拒绝接受这个任务。（实际上抛出RejectedExecutionException异常的操作，是ThreadPoolExecutor线程池中一个默认的RejectedExecutionHandler实现：AbortPolicy）
2、一旦线程池中某个线程完成了任务的执行，它就会试图到任务等待队列中拿去下一个等待任务（所有的等待任务都实现了BlockingQueue接口，按照接口字面上的理解，这是一个可阻塞的队列接口），它会调用等待队列的poll()方法，并停留在哪里。
3、当线程池中的线程超过你设置的corePoolSize参数，说明当前线程池中有所谓的“非核心线程”。那么当某个线程处理完任务后，如果等待keepAliveTime时间后仍然没有新的任务分配给它，那么这个线程将会被回收。线程池回收线程时，对所谓的“核心线程”和“非核心线程”是一视同仁的，直到线程池中线程的数量等于你设置的corePoolSize参数时，回收过程才会停止。

这里关于线程池的具体实现先留个坑，到时候再填。

5. 线程池管理工具 Executors

Executors是线程池创建和使用的工具类，它的所有方法都是static的。

  • 生成一个固定大小的线程池：

public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}

最大线程数设置为与核心线程数相等，此时 keepAliveTime 设置为 0（因为这里它是没用的，即使不为 0，线程池默认也不会回收 corePoolSize 内的线程），任务队列采用 LinkedBlockingQueue，无界队列。

过程分析：刚开始，每提交一个任务都创建一个 worker，当 worker 的数量达到 nThreads 后，不再创建新的线程，而是把任务提交到 LinkedBlockingQueue 中，而且之后线程数始终为 nThreads。

  • 生成只有一个线程的固定线程池，这个更简单，和上面的一样，只要设置线程数为 1 就可以了：

public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}

  • 生成一个需要的时候就创建新的线程，同时可以复用之前创建的线程（如果这个线程当前没有任务）的线程池：

public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}

核心线程数为 0，最大线程数为 Integer.MAX_VALUE，keepAliveTime 为 60 秒，任务队列采用 SynchronousQueue。

这种线程池对于任务可以比较快速地完成的情况有比较好的性能。如果线程空闲了 60 秒都没有任务，那么将关闭此线程并从线程池中移除。所以如果线程池空闲了很长时间也不会有问题，因为随着所有的线程都会被关闭，整个线程池不会占用任何的系统资源。

过程分析：我把 execute 方法的主体黏贴过来，让大家看得明白些。鉴于 corePoolSize 是 0，那么提交任务的时候，直接将任务提交到队列中，由于采用了 SynchronousQueue，所以如果是第一个任务提交的时候，offer 方法肯定会返回 false，因为此时没有任何 worker 对这个任务进行接收，那么将进入到最后一个分支来创建第一个 worker。之后再提交任务的话，取决于是否有空闲下来的线程对任务进行接收，如果有，会进入到第二个 if 语句块中，否则就是和第一个任务一样，进到最后的 else if 分支。

参考：

https://www.cnblogs.com/MOBIN/p/5436482.html

ExectorService 中invokeeAll 接口说明： https://blog.csdn.net/baidu_23086307/article/details/51740852

线程池使用： https://blog.csdn.net/qq_25806863/article/details/71126867

https://javadoop.com/2017/09/05/java-thread-pool/#toc6

https://blog.csdn.net/lipc_/article/details/52025993