Java基础-线程-线程池

2021-02-08 本文已影响0人 HughJin

Java工程师知识树 / Java基础

为什么使用线程池

池化技术相比大家已经屡见不鲜了，线程池、数据库连接池、Http 连接池等等都是对这个思想的应用。池化技术的思想主要是为了减少每次获取资源的消耗，提高对资源的利用率。

线程池提供了一种限制和管理资源（包括执行一个任务）。每个线程池还维护一些基本统计信息，例如已完成任务的数量。

使用线程池的好处：

降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
提高响应速度。当任务到达时，任务可以不需要的等到线程创建就能立即执行。
提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

四种线程池

线程池创建最终是通过ThreadPoolExecutor创建的。根据构造方法传参数的不同又可以分为不同的线程池。

ThreadPoolExecutor 类图结构：

ThreadPoolExecutor构造方法为：

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

参数说明

corePoolSize：线程池中核心线程数的数量
maximumPoolSize：在线程池中允许存在的最大线程数
keepAliveTime：当存在的线程数大于corePoolSize，那么会找到空闲线程去销毁，此参数是设置空闲多久的线程才被销毁。
unit：时间单位 TimeUnit工具类
workQueue：工作队列，线程池中的当前线程数大于核心线程的话，那么接下来的任务会放入到队列中
- ArrayBlockingQueue：基于数组有界阻塞队列
- LinkedBlockingQueue：基于链表阻塞队列
- SynchronousQueue：不存储元素的阻塞队列（读写须等待一并进行）
- PriorityBlockingQueue：支持优先级的无界队列
threadFactory：在创建线程的时候，通过工厂模式来生产线程。这个参数就是设置我们自定义的线程创建工厂。
handler：拒绝策略，如果超过了最大线程数，那么就会执行我们设置的拒绝策略
- AbortPolicy：直接抛出异常。
- CallerRunsPolicy：只用调用者所在线程来运行任务。
- DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
- DiscardPolicy：不处理，丢弃掉。

线程池处理逻辑：

前corePoolSize个任务时，来一个任务就创建一个线程
如果当前线程池的线程数大于了corePoolSize那么接下来再来的任务就会放入到我们上面设置的workQueue队列中
如果此时workQueue也满了，那么再来任务时，就会新建临时线程，那么此时如果我们设置了keepAliveTime或者设置了allowCoreThreadTimeOut，那么系统就会进行线程的活性检查，一旦超时便销毁线程
如果此时线程池中的当前线程大于了maximumPoolSize最大线程数，那么就会执行我们刚才设置的handler拒绝策略

线程池分类：

newFixedThreadPool---固定大小的线程池

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

使用newFixedThreadPool方法创建出来的线程池为固定大小的线程池，可以通过第一个静态方法指定线程池的大小，该线程池corePoolSize和maximumPoolSize相等，阻塞队列使用的是LinkedBlockingQueue，理论上大小为整数最大值。

使用newFixedThreadPool方法创建出来的线程池中的线程数量始终不变，当有新任务提交时，线程池中有空闲线程则会立即执行，如果没有，则会暂存到阻塞队列。

对于固定大小的线程池，不存在线程数量的变，同时使用无界的LinkedBlockingQueue来存放执行的任务。

存在的问题：

当任务提交十分频繁的时候，LinkedBlockingQueue迅速增大，存在着耗尽系统资源的问题。而且在线程池空闲时，即线程池中没有可运行任务时，它也不会释放工作线程，还会占用一定的系统资源，需要shutdown。

ExecutorService fixPool = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
    fixPool.execute(() -> System.out.println(Thread.currentThread().getName() + " : " + atomicInteger.getAndIncrement()));
}
fixPool.shutdown();
// 打印结果 5个线程
pool-1-thread-1 : 0
pool-1-thread-5 : 4
pool-1-thread-4 : 3
pool-1-thread-3 : 2
pool-1-thread-2 : 1
pool-1-thread-3 : 8
pool-1-thread-4 : 7
pool-1-thread-5 : 6
pool-1-thread-1 : 5
pool-1-thread-2 : 9

newSingleThreadExecutor---单个线程的线程池

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

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

使用newSingleThreadExecutor方法创建的线程池为单个线程线程池，只有一个线程的线程池，阻塞队列使用的是LinkedBlockingQueue,若有多余的任务提交到线程池中，则会被暂存到阻塞队列，待空闲时再去执行。按照先入先出的顺序执行任务。

ExecutorService singlePool = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
    singlePool.execute(() -> System.out.println(Thread.currentThread().getName() + " : " + atomicInteger.getAndIncrement()));
}
singlePool.shutdown();
// 打印结果  只有一个线程
pool-1-thread-1 : 0
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 : 5
pool-1-thread-1 : 6
pool-1-thread-1 : 7
pool-1-thread-1 : 8
pool-1-thread-1 : 9

newCachedThreadPool---缓存线程池

java.util.concurrent.Executors
    
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS, // 缓存的线程默认存活时间
                                  new SynchronousQueue<Runnable>());
}

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,// 缓存的线程默认存活时间
                                  new SynchronousQueue<Runnable>(),
                                  threadFactory);
}

使用newCachedThreadPool方法创建的线程池为缓存线程池。

缓存线程池，缓存的线程默认存活60秒。线程的核心池corePoolSize大小为0，核心池最大为Integer.MAX_VALUE,阻塞队列使用的是SynchronousQueue。

SynchronousQueue是一个直接提交的阻塞队列， SynchronousQueue总会迫使线程池增加新的线程去执行新的任务。在没有任务执行时，当线程的空闲时间超过keepAliveTime（60秒），则工作线程将会终止被回收，当提交新任务时，如果没有空闲线程，则创建新线程执行任务，会导致一定的系统开销。

存在的问题：

如果同时又大量任务被提交，而且任务执行的时间不是特别快，那么线程池便会新增出等量的线程池处理任务，这很可能会很快耗尽系统的资源。

ExecutorService cachedPool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
    cachedPool.execute(() -> System.out.println(Thread.currentThread().getName() + " : " + atomicInteger.getAndIncrement()));
}
//打印结果
pool-1-thread-1 : 0
pool-1-thread-4 : 3
pool-1-thread-3 : 2
pool-1-thread-2 : 1
pool-1-thread-6 : 5
pool-1-thread-5 : 4
pool-1-thread-7 : 6
pool-1-thread-8 : 7
pool-1-thread-8 : 8     
pool-1-thread-2 : 9

newScheduledThreadPool---定时线程池

java.util.concurrent.Executors
    
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

使用newScheduledThreadPool方法创建的线程池为定时线程池。

定时线程池，可用于周期性地去执行任务，通常用于周期性的同步数据。

ScheduledExecutorService ses = Executors.newScheduledThreadPool(5);
//5个线程每两秒执行一次 随机哪个线程执行
ses.scheduleAtFixedRate(() -> System.out.println(Thread.currentThread().getName() + " : " + "执行了"), 0, 2, TimeUnit.SECONDS);
打印结果:  5个线程每两秒执行一次 随机哪个线程执行
pool-1-thread-1 : 执行了
pool-1-thread-1 : 执行了
pool-1-thread-2 : 执行了
pool-1-thread-1 : 执行了
pool-1-thread-3 : 执行了
pool-1-thread-3 : 执行了
pool-1-thread-3 : 执行了
... ...

线程池的使用

线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。

说明：Executors各个方法的弊端：

1：newFixedThreadPool和newSingleThreadExecutor:
主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。

2：newCachedThreadPool和newScheduledThreadPool:
主要问题是线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM。

使用实例：

package com.thread.study.pool;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadTest {

    public static void main(String[] args) throws InterruptedException {
        int corePoolSize = 3; // 线程池中核心线程数的数量
        int maximumPoolSize = 6;// 在线程池中允许存在的最大线程数
        long keepAliveTime = 10;// 存活时间
        TimeUnit unit = TimeUnit.SECONDS; // 时间单位 秒
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(3); // 工作队列 基于链表阻塞队列
        ThreadFactory threadFactory = new NameTreadFactory();// 自定义线程工厂
        RejectedExecutionHandler handler = new MyIgnorePolicy(); // 自定义拒绝策略
        /*
            public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                              long keepAliveTime, TimeUnit unit,
                              BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
         */
        ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                workQueue, threadFactory, handler);
        executor.prestartAllCoreThreads(); // 预启动所有核心线程
        for (int i = 1; i <= 10; i++) {
            executor.execute(new MyTask(String.valueOf(i)));
        }
        System.out.println("当前线程池中存活的线程数为: "+executor.getActiveCount());
        Thread.sleep(5000);
        System.out.println("当前线程池中存活的线程数为: "+executor.getActiveCount());
        executor.shutdown();
    }

    // 自定义线程工厂
    static class NameTreadFactory implements ThreadFactory {

        private final AtomicInteger mThreadNum = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "my-thread-" + mThreadNum.getAndIncrement());
            System.out.println(t.getName() + " has been created");
            return t;
        }
    }

    //拒绝策略
    public static class MyIgnorePolicy implements RejectedExecutionHandler {

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            doLog(r, e);
        }

        private void doLog(Runnable r, ThreadPoolExecutor e) {
            // 可做日志记录等
            System.err.println(r.toString() + " rejected!" + " 当前线程池中存活的线程数为: " + e.getActiveCount());
        }
    }

    // 线程
    static class MyTask implements Runnable {
        private String name;

        public MyTask(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            try {
                System.out.println(Thread.currentThread().getName() +" : "+ this.toString() + " is running!");
                Thread.sleep(1000); //让任务执行慢点
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public String getName() {
            return name;
        }

        @Override
        public String toString() {
            return "MyTask [name=" + name + "]";
        }
    }
}

打印结果：

my-thread-1 has been created
my-thread-2 has been created
my-thread-3 has been created
my-thread-4 has been created
my-thread-3 : MyTask [name=3] is running!
my-thread-1 : MyTask [name=1] is running!
my-thread-2 : MyTask [name=2] is running!
my-thread-5 has been created
my-thread-6 has been created
my-thread-4 : MyTask [name=4] is running!
my-thread-5 : MyTask [name=8] is running!
my-thread-6 : MyTask [name=9] is running!
当前线程池中存活的线程数为: 6
MyTask [name=10] rejected! 当前线程池中存活的线程数为: 6
my-thread-3 : MyTask [name=7] is running!
my-thread-2 : MyTask [name=5] is running!
my-thread-1 : MyTask [name=6] is running!
当前线程池中存活的线程数为: 0

线程池使用场景

使用线程池异步操作常见场景

批量操作
批量操作不是实时显示效果的操作，比如批量导入配置，批量删除或作废，批量导出查询结果等，只需要搭建一个任务，在任务中处理即可，有错误对应处理错误

日志
异步写操作对功能影响不大的业务逻辑是常见的场景，日志对与功能来说重要性并么有那么高

使用第三方接口
比如发邮箱，发短信，发消息队列信息，推送搜索引擎数据，异步调用外围接口等处理数据。