Java 使用线程池

目录

一.介绍

二.ThreadPoolExecutor

　　2.1 认识ThreadPoolExecutor

　　2.2 ThreadPoolExecutor的构造方法列表

　　2.3 任务队列分类

　　2.4 线程工厂

　　2.5 拒绝策略

三.使用Executors快速创建线程池

 

一.介绍

　　本文主要介绍线程池的创建、线程任务队列分类以及线程池的扩展相关内容，只包含比较基础的知识，可以用来复习知识。

二.ThreadPoolExecutor

2.1 认识ThreadPoolExecutor

　　ThreadPoolExecutor，可以理解为ThreadPool+Executor（线程池+执行器），执行器会使用线程池中的线程来处理提交的任务，可以看下面的继承关系图：

　　

2.2 ThreadPoolExecutor的构造方法列表

　　创建线程池，就是创建ThreadPoolExecutor实例对象，需要注意的是ThreadPoolExecutor有多个构造方法，如下所示：

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

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

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

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

　　corePoolSize：核心线程数（作用相当于一个门限值，看后面的介绍就知道了）

　　maximumPoolSize：线程池中线程的最大数量；

keepAliveTime，TimeUnit：当线程池中的线程数量超过corePoolSize后，多余的空闲线程存活的时间，也就是超过corePoolSize的空闲线程在多长时间后会被销毁；

　　workQueue：任务队列，用来保存已经提交但是并没有被执行的任务；

　　threadFactory：线程池中线程的创建工厂（默认使用Executors.DefaultThreadFactory）；

　　handler：拒绝策略，当任务来不及被处理时被拒绝的处理操作；

2.3 任务队列分类

SynchronousQueue（同步队列）：

1.当前线程池中的线程数量没有达到最大线程数，当任务被提交时：

a.如果有空闲线程，则让空闲线程来处理任务；

b.如果没有空闲线程，那么就会创建新的线程来处理任务；

2.当前线程池中的线程数量已经达到最大线程数，当任务被提交时：

a.如果有空闲线程，则让空闲线程来处理任务；

b.如果没有空闲线程，那么就会执行拒绝策略；这个时候，如果希望任务能被提交，那么就必须等待一个线程执行完成，空出一个空闲线程，所以“每一个插入操作，都必须等待一个删除操作”。

3.综上，SynchronousQueue没有容量的概念，只有maximumPoolSize起着“容量”的作用。

ArrayBlockingQueue（有界队列）

1.创建ArrayBlockingQueue必须指定容量，也就是队列的最大长度；

2.当有新任务提交的时候，会有下面几种情况：

a.如果线程池中线程的数量小于corePoolSize，那么就会创建新线程来处理任务（如果有空闲线程，则让空闲线程处理）；

b.如果线程池中的线程数量达到corePoolSize，那么就会将任务提交到任务队列；

c.如果任务队列已经满了，那么就会在线程池中数量不超过maximumPoolSize的前提下，创建线程来处理提交的任务；

d.如果任务队列已满，且线程池中线程数量已经达到maximumPoolSize数量，那么就会执行拒绝策略；

3.综上，使用ArrayBlockingQueue，线程池只有当任务队列满了之后才会创建超过corePoolSize的线程进行处理任务，在这之前，线程池中的最多保持corePoolSize个线程；

LinkedBlockingQueue（无界队列）

1.创建LinkedBlockingQueue时可以指定容量，也可以不指定容量；

2.对于不指定容量的LinkedBlockingQueue，当有任务提交的时候：

a.如果线程池中的线程数量小于corePoolSize，那么就会创建新线程来执行任务（如果有空闲线程，则让空闲线程处理）；

b.如果线程池中的线程数量达到corePoolSize后，如果有空闲线程，那么就让空闲线程来执行任务，如果没有空闲线程，那么就会将任务提交到任务队列，而不会创建新线程来执行任务；

3.综上，使用LinkedBlockingQueue时，线程数量达到corePoolSize后，就不会再创建新线程了，后面的任务都会直接进入任务队列，如此的话，如果提交的任务很多，则会造成系统资源枯竭；

4.如果担心任务无线提交，则可以创建LinkedBlockingQueue时指定队列容量，超过队列容量后，再往队列中提交任务，则会触发拒绝策略；

 

PropertyBlockingQueue（优先队列）

为任务设置优先级，提交到任务队列后，会根据优先级进行处理任务，而不是按照先到先服务的算法进行处理（SynchronousQueue、ArrayBlockingQueue、LinkedBlockingQueue中的任务都是按照FIFO算法处理）。

2.4 线程工厂

　　熟悉设计模式的话，就知道XX工厂，就是用来创建XX的，所以ThreadFactory就是用来创建线程的，一般来说，我们直接使用默认的线程工厂即可，也就是Executors.DefaultThreadFactory；

　　但是有时候我们需要对线程的创建做一些扩展，就可以通过对创建自己的线程工厂来实现，可以参考这篇博客，利用线程自定义线程工厂来对线程池中的线程进行命名的方式：Java的每个Thread都希望拥有自己的名称

　　注意，可以通过继承ThreadPoolFactory来扩展线程池。

2.5 拒绝策略

　　当任务不能提交到任务队列时，执行的操作，称为拒绝策略，只需要实现RejectedExecutionHandler接口接口：

package cn.ganlixin.thread;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public class MyRejectedExecutionHandler implements RejectedExecutionHandler {

    /**
     * 拒绝策略处理逻辑
     *
     * @param task     提交到任务队列被拒绝的任务
     * @param executor 执行任务的执行器（线程池）
     */
    @Override
    public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
        // 在这里定义逻辑，我这里只是输出一串文字
        log.error("提交任务被拒绝");
    }
}

　　

　　使用示例：

package cn.ganlixin.thread;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

@Slf4j
public class TestThread {

    public static void main(String[] args) throws InterruptedException {

        Runnable runnable = () -> {
            Thread thread = Thread.currentThread();
            String name = thread.getName();
            log.info("thread {} running", name);

            try {
                TimeUnit.SECONDS.sleep(5); // 休眠5秒
            } catch (InterruptedException ignored) {
            }
        };

        int corePoolSize = 2;
        int maximumPoolSize = 4; // 最多4个线程
        int keepAliveSecond = 5;
        ExecutorService executorService = new ThreadPoolExecutor(
                corePoolSize, maximumPoolSize, keepAliveSecond, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2), // 任务队列容量为2
                Executors.defaultThreadFactory(),
                new MyRejectedExecutionHandler());

        // 提交8个任务，因为线程池最多有4线程，再加上2个任务队列的任务，所以有2个任务会被拒绝
        for (int i = 0; i < 8; i++) {
            executorService.submit(runnable);
        }
        
        // 输出如下：
        // thread pool-2-thread-3 running
        // thread pool-2-thread-2 running
        // thread pool-2-thread-4 running
        // thread pool-2-thread-1 running
        // 提交任务被拒绝
        // 提交任务被拒绝
        // thread pool-2-thread-4 running
        // thread pool-2-thread-2 running
        
        Thread.currentThread().join();
    }
}

　　

三.使用Executors快速创建线程池

　　可以利用Executors类的各个静态方法来创建线程池，创建的线程池特点从名称上就可以看出来，需要注意的这些静态方法内部其实也是ThreadPoolFactory实例对象，也就是线程池，只不过根据不同的类型设置了不同的参数。