看了就能学会的Java线程池技术

Posted 活跃的咸鱼

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了看了就能学会的Java线程池技术相关的知识，希望对你有一定的参考价值。

线程池

1.线程池的介绍

1.1 线程池的概念

线程池：Java中开辟出了一种管理线程的概念，这个概念叫做线程池，使用池的技术可以降低资源消耗（常见的池技术有jdbc连接池，线程池，内存池，对象池）。

1.2 为什么要提出线程池的概念

程序运行的本质是cpu进程的调度（占用系统的资源）进程是一个动态的过程，是一个活动的实体。简单来说，一个应用程序的运行就可以被看做是一个进程，而线程，是运行中的实际的任务执行者。可以说，进程中包含了多个可以同时运行的线程。而在Java中，内存资源是极其宝贵的，所以，我们就提出了线程池的概念。

1.3 线程池的好处

（1）降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。

（2）提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。

（3）提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

2. 线程池的使用

2.1 线程池的创建

我们可以通过ThreadPoolExecutor来创建一个线程池。
ThreadPoolExecutor类是线程池的核心实现类，用来执行被提交的任务。其继承关系如下。
在这里插入图片描述

下面我们来分析一下ThreadPoolExecutor类我们来看看ThreadPoolExecutor类的构造方法：

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }


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

从源码中我们发现ThreadPoolExecutor有四个构造方法，所以有四种创建方法，在四个构造函数中一共有七个参数下面来对七个参数进行说明。

2.2 线程池的七大参数

(1)、corePoolSize（核心线程数，必需指定）：

线程池的基本大小。当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于基本线程时就不再创建。如果调用了线程池的prestartAllCoreThreads()方法，线程池会提前将核心线程创建并启动。默认情况下，核心线程会一直存活，但是当将 allowCoreThreadTimeout 设置为 true 时，核心线程也会超时回收。

(2)、maximumPoolSize（必需指定，线程池最大数量）：

线程池所能容纳的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会创建新的线程执行任务。不过要注意，如果使用了无界的任务队列这个参数则没有任何效果，因为线程创建要等到队列满了才会创建。

(3)、keepAliveTime（必需指定，线程活动保持的最长时间）：

线程池的工作线程空闲后，保持的存活时间，所以如果任务很多，并且每个任务执行的时间比较短，可以调大时间，提高线程的利用率。如果超过该时长，非核心线程就会被回收。如果将allowCoreThreadTimeout 设置为 true 时，核心线程也会超时回收。

(4)、TimeUnit（必需指定）：

指定keepAliveTime 参数的时间单位。常用的有：TimeUnit.MILLISECONDS（毫秒）、TimeUnit.SECONDS（秒）、TimeUnit.MINUTES（分）。

（5)、BlockingQueue （任务队列，必需指定）：

用于保存等待执行的任务的阻塞队列，通过线程池的 execute() 方法提交的 Runnable 对象将存储在该参数中。

阻塞队列的阻塞分两种情况：

  1. 当队列满的时候去写会阻塞
  2. 当队列为空的时候去读也会阻塞

BlockingQueue：四组API

功能抛出异常不抛出异常且有返回值阻塞等待超时等待
添加add()offer()put()offer(…)
移除remove()poll()take()poll(…)
检查队首元素element()peek()--
//抛异常
 public static void test1(){
        System.out.println(queue.add("a"));
        System.out.println(queue.add("b"));
        System.out.println(queue.add("c"));
        //队列满 抛出IllegalStateException
        System.out.println(queue.add("d"));
        System.out.println("---------------------");
        System.out.println(queue.remove());
        System.out.println(queue.remove());
        System.out.println(queue.remove());
         //如果队列为空 NoSuchElementException
        System.out.println(queue.remove());
        //返回队首的元素
        System.out.println(queue.element());
    }
    
  //不抛异常，成功则会返回true失败返回false
  public static void test2(){
        
        System.out.println(queue.offer("d"));
        System.out.println(queue.offer("d"));
        System.out.println(queue.offer("d"));
        //队列满不抛异常会返回false
        System.out.println(queue.offer("d"));

        System.out.println(queue.poll());
        System.out.println(queue.poll());
        System.out.println(queue.poll());
        //没有元素弹出会返回null
        System.out.println(queue.poll());
        //返回队首的元素
        System.out.println(queue.peek());

    }
//等待阻塞（一直阻塞）
public static void test3()throws Exception{
      
        queue.put("a");
        queue.put("b");
        queue.put("c");
        //阻塞等待一直阻塞
        queue.put("d");
        queue.take();
        queue.take();
        queue.take();
        //没有元素也会一直阻塞
        queue.take();
    }
  //等待超时退出
 public static void test4()throws InterruptedException{
        System.out.println(queue.offer("d"));
        System.out.println(queue.offer("d"));
        System.out.println(queue.offer("d"));
        //阻塞两秒后自动退出
        System.out.println(queue.offer("d",2,TimeUnit.SECONDS));
        System.out.println(queue.poll(2, TimeUnit.SECONDS));
        System.out.println(queue.poll(2, TimeUnit.SECONDS));
        System.out.println(queue.poll(2, TimeUnit.SECONDS));
        //阻塞两秒后自动退出
        System.out.println(queue.poll(2, TimeUnit.SECONDS));
    }

任务队列是基于阻塞队列实现的，即采用生产者消费者模式，在 Java 中需要实现 BlockingQueue 接口。该接口的实现子类如下：
在这里插入图片描述
ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列（数组结构可配合指针实现一个环形队列），此队列按照先进先出原则对元素进行排序。

LinkedBlockingQueue： 一个由链表结构组成的有界阻塞队列，也是按照先进先出排序元素，吞吐量要高于ArrayBlockingQueue，静态工厂方法Executors.newFixedThreadPool使用了这个队列。在未指明容量时，容量默认为 Integer.MAX_VALUE。

PriorityBlockingQueue： 一个支持优先级排序的无界阻塞队列，对元素没有要求，可以实现 Comparable 接口也可以提供 Comparator 来对队列中的元素进行比较。跟时间没有任何关系，仅仅是按照优先级取任务。

DelayQueue：类似于PriorityBlockingQueue，是二叉堆实现的无界优先级阻塞队列。要求元素都实现 Delayed 接口，通过执行时延从队列中提取任务，时间没到任务取不出来。

SynchronousQueue： 一个不存储元素的阻塞队列，消费者线程调用 take() 方法的时候就会发生阻塞，直到有一个生产者线程生产了一个元素，消费者线程就可以拿到这个元素并返回；生产者线程调用 put() 方法的时候也会发生阻塞，直到有一个消费者线程消费了一个元素，生产者才会返回。

public class TestSynchronousQueue {
    public static void main(String[] args) {
        BlockingQueue<String> queue=new SynchronousQueue<>();
        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+" put a");
                queue.put("a");
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+" put b");
                queue.put("b");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"product").start();

        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"=>"+ queue.take());
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"=>"+ queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"consumer").start();
    }
}

结果：
product put a
consumer=>a
product put b
consumer=>b

LinkedBlockingDeque： 使用双向队列实现的有界双端阻塞队列。双端意味着可以像普通队列一样 FIFO（先进先出），也可以像栈一样 FILO（先进后出）。

LinkedTransferQueue： 它是ConcurrentLinkedQueue、LinkedBlockingQueue 和 SynchronousQueue 的结合体，但是把它用在 ThreadPoolExecutor 中，和 LinkedBlockingQueue 行为一致，但是是无界的阻塞队列。

注意：

有界队列和无界队列的区别：如果使用有界队列，当队列饱和时并超过最大线程数时就会执行拒绝策略；而如果使用无界队列，因为任务队列永远都可以添加任务，所以设置maximumPoolSize 没有任何意义。

(6)、ThreadFactory（线程工厂，可选）：用于指定为线程池创建新线程的方式，还可以为每个创建出来的线程设置线程名。使用开源框架guava提供的ThreadFactoryBuilder可以快速的设置。

new ThreadFactoryBuilder().setNameFormat("XX-task-%d").build();

(7）、 RejectedExecutionHandler（拒绝策略，可选）：当达到最大线程数时需要执行的饱和策略。

当线程池的线程数达到最大线程数时，需要执行拒绝策略。拒绝策略需要实现 RejectedExecutionHandler 接口，并实现 rejectedExecution(Runnable r, ThreadPoolExecutor executor) 方法。在ThreadPoolExecutor类中实现了 4 种拒绝策略：
在这里插入图片描述

AbortPolicy（默认）：丢弃任务并抛出 RejectedExecutionException 异常。

public class Demo1 {
    public static void main(String[] args) {
        //创建一个核心线程为2，最大容量线程为5线程闲置超时时长为3秒，采取拒绝策略的线程池
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2, 5, 3, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(3), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
        //最大承载=最大容量线程+队列数量=8
        //9>8超过最大承重量报RejectedExecutionException
        for (int i = 0; i < 9; i++) {
           poolExecutor.execute(()->{
                System.out.println(Thread.currentThread().getName());
            });
        }

    }
}
测试结果：
pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-4
pool-1-thread-3
pool-1-thread-1
pool-1-thread-5
pool-1-thread-2
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task threadpool.demo2$$Lambda$14/0x0000000100066840@71be98f5 rejected from java.util.concurrent.ThreadPoolExecutor@6fadae5d[Running, pool size = 5, active threads = 5, queued tasks = 0, completed tasks = 3]
	at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2055)
	at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:825)
	at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1355)
	at threadpool.demo2.main(demo2.java:16)

CallerRunsPolicy：由调用线程处理该任务。

public class Demo2 {
    public static void main(String[] args) {
        //创建一个核心线程为2，最大容量线程为5线程闲置超时时长为3秒
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2, 5, 3, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(3), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());
        //最大承载=最大容量线程+队列数量
        //超过由调用线程处理该任务。
        for (int i = 0; i < 9; i++) {
           poolExecutor.execute(()->{
                System.out.println(Thread.currentThread().getName());
            });
        }

    }
}
超过了由主线程来处理

pool-1-thread-1
pool-1-thread-3
main
pool-1-thread-2
pool-1-thread-2
pool-1-thread-2
pool-1-thread-2
pool-1-thread-4
pool-1-thread-5

DiscardPolicy：丢弃任务，但是不抛出异常。可以配合这种模式进行自定义的处理方式。

public class Demo3 {
    public static void 
以上是关于看了就能学会的Java线程池技术的主要内容，如果未能解决你的问题，请参考以下文章
 

 Java线程池详解


 Java-----关于线程池的使用


 你真的懂ThreadPoolExecutor线程池技术吗？看了源码你会有全新的认识


 看了就很快学会jQuery


 史前最详细的：Java线程池核心流程解析


 java线程池的初探