zoukankan      html  css  js  c++  java

  • 一文弄懂Java的线程池 原创

    https://blog.csdn.net/Kurry4ever_/article/details/109294661

    文章目录

    前言

    工作中难免会使用线程池。对线程池的使用要格外的小心，说不定某天就出现了难搞的生产问题（OOM）。每次在使用的时候，我都会网上找找资料，今天我就自己整理这篇文章，不足或错误之处，希望大家看完后多多补充，提提意见。


    1、为什么要使用多线程

      我们使用多线程的本质是为了提升程序的性能。程序的性能我们可以用2个指标来度量：

    • 延迟：发出请求到收到响应这个过程的时间；延迟越短，意味着程序执行得越快，性能也就越好。
    • 吞吐量：在单位时间内能处理请求的数量；吞吐量越大，意味着程序能处理的请求越多，性能也就越好。
      同等条件下，延迟越短，吞吐量越大。但是由于它们隶属不同的维度（一个是时间维度，一个是空间维度），并不能互相转换。
      我们所谓提升性能，从度量的角度，主要是降低延迟，提高吞吐量。

      要想“降低延迟，提高吞吐量”，对应的方法呢，基本上有两个方向，一个方向是优化算法，另一个方向是将硬件的性能发挥到极致。前者属于算法范畴，后者则是和并发编程息息相关了。那计算机主要有哪些硬件呢？主要是两类：一个是 I/O，一个是 CPU。简言之，在并发编程领域，提升性能本质上就是提升硬件的利用率，再具体点来说，就是提升 I/O 的利用率和 CPU 的利用率。

      在单核时代，多线程主要就是用来平衡 CPU 和 I/O 设备的。如果程序只有 CPU 计算，而没有 I/O 操作的话，多线程不但不会提升性能，还会使性能变得更差，原因是增加了线程切换的成本。但是在多核时代，这种纯计算型的程序也可以利用多线程来提升性能。为什么呢？因为利用多核可以降低响应时间。



    2、创建线程几种方式

    继承Thread类：Thread是类，有单继承的局限性。
    实现Runnable接口：任务和线程分开，不能返回执行结果。
    实现Callable接口：利用FutureTask执行任务，能取到执行结果。

    package com.top.test.threads;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class ThreadPoolTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Thread a = new A();
        Thread b = new Thread(new B());
        FutureTask<String> futureTask = new FutureTask<>(new C());
        Thread c = new Thread(futureTask);

        a.start();
        b.start();
        c.start();
        System.out.println(futureTask.get());
    }
}

class A extends Thread {
    @Override
    public void run() {
        System.out.println("继承Thread的线程任务");
    }
}

class B implements Runnable {
    @Override
    public void run() {
        System.out.println("实现Runnable的线程任务");
    }
}

class C implements Callable<String> {
    @Override
    public String call() throws Exception {
        return "实现Callable的线程任务";
    }
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41

      但是我们工作中一般不这样来创建线程。原因：虽然在 Java 语言中创建线程看上去就像创建一个对象一样简单，只需要 new Thread() 就可以了，但实际上创建线程远不是创建一个对象那么简单。创建对象，仅仅是在 JVM 的堆里分配一块内存而已；而创建一个线程，却需要调用操作系统内核的 API，然后操作系统要为线程分配一系列的资源，这个成本就很高了，所以线程是一个重量级的对象，应该避免频繁创建和销毁。

      那应该怎样创建线程呢？你应该立刻想到了用线程池。利用线程池把资源池化，使得线程资源能服用，可以避免频繁地创建和销毁。



    3、生产者-消费者模式

      线程池的设计采用了生产者-消费者模式。线程池的使用方是生产者，线程池本身是消费者，阻塞队列来存储要处理的任务。简单画了个图：

    在这里插入图片描述



    4、Java中的线程池

      从jdk1.5版本开始，在java.uitl.concurrent包下面定义定义了一些列与并发相关的类，其中线程池最核心的一个类是ThreadPoolExecutor。

      查看ThreadPoolExecutor的源码，看下基本的继承关系：

    public class ThreadPoolExecutor extends AbstractExecutorService {
	…
}

public abstract class AbstractExecutorService implements ExecutorService {
	…
}
public interface ExecutorService extends Executor {
	…
}
public interface Executor {
    void execute(Runnable command);
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

      我们可以看出，Executor接口中定义了execute方法，execute是用来执行我们提交的任务的。

      但是类ThreadPoolExecutor源码注释中，是推荐我们使用类Executors的工程方法来创建线程池的：

    * <p>To be useful across a wide range of contexts, this class
* provides many adjustable parameters and extensibility
* hooks. However, programmers are urged to use the more convenient
* {@link Executors} factory methods {@link
* Executors#newCachedThreadPool} (unbounded thread pool, with
* automatic thread reclamation), {@link Executors#newFixedThreadPool}
* (fixed size thread pool) and {@link
* Executors#newSingleThreadExecutor} (single background thread), that
* preconfigure settings for the most common usage
* scenarios.
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    看下源码，Executors提供四种线程池，分别为：

    • newCachedThreadPool：创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
    public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • newFixedThreadPool ：创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
    public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 1
    • 2
    • 3
    • 4
    • 5
    • newScheduledThreadPool ：创建一个定长线程池，支持定时及周期性任务执行。
    public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}
     
    • 1
    • 2
    • 3
    • 4
    • 1
    • 2
    • 3
    • 4
    • newSingleThreadExecutor：创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
    public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    对于上面创建四种线程池，我这里就不一一写例子了，网上多的是例子。


      既然JDK提供了这么好的工具类，我们是不是就肯定选择它呢？并不是，在阿里开发手册中有这样一条：

    在这里插入图片描述

    需要阿里开发手册的同学，可在公众号“Java尖子生”，回复“alibaba”领取。

      看来，最终都有可能导致OOM，而 OOM 会导致所有请求都无法处理，这是致命问题。所以强烈建议使用有界队列，并设置最大线程数。



    5、自定义线程池ThreadPoolExecutor

      从上面的源码，我们也看到的四种线程池的最终方式也是调用的ThreadPoolExecutor的构造方法：

    public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                          long keepAliveTime, TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

      ThreadPoolExecutor 的构造函数非常复杂，现在我们自己通过ThreadPoolExecutor来创建线程池，那么我们就有必要详细了解下其构造函数的每一个参数。

    5.1 corePoolSize

      核心池（核心线程数）的大小。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法的名字就可以看出，是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列（阻塞队列）当中。
    核心线程数的选择很重要，那创建多少个线程是合理的呢？

    • 对于 CPU 密集型的计算场景，理论上“线程的数量 =CPU 核数”就是最合适的。不过在工程上，线程的数量一般会设置为“CPU 核数 +1”，这样的话，当线程因为偶尔的内存页失效或其他原因导致阻塞时，这个额外的线程可以顶上，从而保证 CPU 的利用率。

    • 对于 I/O 密集型计算场景，最佳的线程数是与程序中 CPU 计算和 I/O 操作的耗时比相关的，我们可以总结出这样一个公式【理论值】：
      最佳线程数 =CPU 核数 * [ 1 +（I/O 耗时 / CPU 耗时）]

      但实际【经验值】告诉我们应该为：2 * CPU 的核数 + 1

      I/O 耗时和 CPU 耗时的比值是一个关键参数，不幸的是这个参数是未知的，而且是动态变化的，所以工程上，我们要估算这个参数，然后做各种不同场景下的压测来验证我们的估计。不过工程上，原则还是将硬件的性能发挥到极致，所以压测时，我们需要重点关注 CPU、I/O 设备的利用率和性能指标（响应时间、吞吐量）之间的关系。

      工作中都是按照逻辑核数来的，理论值和经验值只是提供个指导，实际上还是要靠压测！！！

    5.2 maximumPoolSize

      线程池最大线程数，表示在线程池中最多能创建多少个线程。当阻塞队列装满时，继续提交任务，会创建救急（非核心）线程来处理。注意： 不是先前创建的线程是核心线程，后面创建的线程是非核心线程，线程是没有核心非核心的概念的。

    5.3 不同场景下提交任务，线程池的表现形式

    其中workerThread表示工作线程数。

    • 当workerThread等于0时：提交任务，创建线程
    • 当workerThread小于corePoolSize时：提交任务，创建线程，不管其他工作线程是不是闲置的。
    • 当workerThread大于等于corePoolSize 且 workerThread小于maxinumPoolsize时：将任务添加到队列中，当队列满了后，创建线程。当一个线程完成任务时，它会从队列中取下一个任务来执行。
    • 当workerThread等于maxinumPoolsize时：提交任务,既不能加入队列，也不能创建新的线程，将RejectedExecutionHandler的rejectedExecution方法执行拒绝策略。

    下面是ThreadPoolExecutor 的execute()方法的源码及注释：

    /**
 * Executes the given task sometime in the future.  The task
 * may execute in a new thread or in an existing pooled thread.
 *
 * If the task cannot be submitted for execution, either because this
 * executor has been shutdown or because its capacity has been reached,
 * the task is handled by the current {@code RejectedExecutionHandler}.
 *
 * @param command the task to execute
 * @throws RejectedExecutionException at discretion of
 *         {@code RejectedExecutionHandler}, if the task
 *         cannot be accepted for execution
 * @throws NullPointerException if {@code command} is null
 */
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}
/**
 * Invokes the rejected execution handler for the given command.
 * Package-protected for use by ScheduledThreadPoolExecutor.
 */
final void reject(Runnable command) {
    handler.rejectedExecution(command, this);
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60

    5.4 keepAliveTime和unit

      keepAliveTime表示线程没有任务执行时最多保持多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0；当一个线程无事可做，超过一定的时间（keepAliveTime）时，线程池会判断，如果当前运行的线程数大于 corePoolSize，那么这个线程就被停掉。所以线程池的所有任务完成后，它最终会收缩到 corePoolSize 的大小。

      unit参数是keepAliveTime的时间单位。

    在TimeUnit类中有7种静态属性：
        TimeUnit.DAYS; //天
        TimeUnit.HOURS; //小时
        TimeUnit.MINUTES; //分钟
        TimeUnit.SECONDS; //秒
        TimeUnit.MILLISECONDS; //毫秒
        TimeUnit.MICROSECONDS; //微妙
        TimeUnit.NANOSECONDS; //纳秒


    5.5 workQueue

      一个阻塞队列，用来存储等待执行的任务，这个参数的选择也很重要，会对线程池的运行过程产生重大影响，一般来说，这里的阻塞队列有以下几种选择：

    • SynchronousQueue
      SynchronousQueue是无界的，是一种无缓冲的等待队列，但是由于该Queue本身的特性，在某次添加元素后必须等待其他线程取走后才能继续添加；可以认为SynchronousQueue是一个缓存值为1的阻塞队列，但是 isEmpty()方法永远返回是true，remainingCapacity() 方法永远返回是0，remove()和removeAll() 方法永远返回是false，iterator()方法永远返回空，peek()方法永远返回null。
      声明一个SynchronousQueue有两种不同的方式，它们之间有着不太一样的行为。公平模式和非公平模式的区别:如果采用公平模式：SynchronousQueue会采用公平锁，并配合一个FIFO队列来阻塞多余的生产者和消费者，从而体系整体的公平策略；但如果是非公平模式（SynchronousQueue默认）：SynchronousQueue采用非公平锁，同时配合一个LIFO队列来管理多余的生产者和消费者，而后一种模式，如果生产者和消费者的处理速度有差距，则很容易出现饥渴的情况，即可能有某些生产者或者是消费者的数据永远都得不到处理。

    • LinkedBlockingQueue
      LinkedBlockingQueue是无界的，是一个无界缓存的等待队列。
      基于链表的阻塞队列，内部维持着一个数据缓冲队列（该队列由链表构成）。当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于同样的原理。
      LinkedBlockingQueue之所以能够高效的处理并发数据，还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。

    • ArrayListBlockingQueue
      ArrayListBlockingQueue是有界的，是一个有界缓存的等待队列。
      基于数组的阻塞队列，同LinkedBlockingQueue类似，内部维持着一个定长数据缓冲队列（该队列由数组构成）。ArrayBlockingQueue内部还保存着两个整形变量，分别标识着队列的头部和尾部在数组中的位置。
      ArrayBlockingQueue在生产者放入数据和消费者获取数据，都是共用同一个锁对象，由此也意味着两者无法真正并行运行，这点尤其不同于LinkedBlockingQueue；按照实现原理来分析，ArrayBlockingQueue完全可以采用分离锁，从而实现生产者和消费者操作的完全并行运行。Doug Lea之所以没这样去做，也许是因为ArrayBlockingQueue的数据写入和获取操作已经足够轻巧，以至于引入独立的锁机制，除了给代码带来额外的复杂性外，其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不同之处在于，前者在插入或删除元素时不会产生或销毁任何额外的对象实例，而后者则会生成一个额外的Node对象。这在长时间内需要高效并发地处理大批量数据的系统中，其对于GC的影响还是存在一定的区别。

    ArrayBlockingQueue和LinkedBlockingQueue是两个最普通、最常用的阻塞队列，一般情况下，处理多线程间的生产者消费者问题，使用这两个类足以。

    关于队列、阻塞队列的知识也很重要，后面我会尽可能的整理一篇文章。


    5.6 threadFactory

      通过这个参数你可以自定义如何创建线程，例如你可以给线程指定一个有意义的名字。

      给线程赋予一个有意义的名字很重要，我这里整理了几种方法：

    • 方法一：自己实现ThreadFactory并制定给线程池，在实现的ThreadFactory中设定计数和调用Thread.setName。
    • 方法二：guava的ThreadFactoryBuilder.setNameFormat可以指定一个前缀，使用%d表示序号；例如：
    ThreadFactory myThreadFactory =(new ThreadFactoryBuilder()).setNameFormat("my-pool-thread-%d").build();
     
    • 1
    • 1

      ThreadFactory是JDK自带的，下面我们详细看下如何用ThreadFactory来实现。

    ThreadFactory是一个只有一个newThread方法的接口：

    public interface ThreadFactory {
    Thread newThread(Runnable r);
}
     
    • 1
    • 2
    • 3
    • 4
    • 1
    • 2
    • 3
    • 4

      我们利用这个方法我们可以做什么呢？可以给线程命名，查看线程数，指定是否守护线程，设置线程优先级等等。

      下面是我写的完整例子，你可以拿这个例子做其他场景的测试（对shutdown等方法不太懂的，往下看）：

    package com.top.test.threads;

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

public class TestThreadFactory {
    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(8, 100,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(100),
                new CustomThreadFactory(false, 1), new ThreadPoolExecutor.AbortPolicy());
        System.out.println("开始往线程池中提交任务");
        // 可以把10改成其他值来测试其他场景哦-_-
        for (int i = 0; i < 10; i++) {
            threadPool.submit(() -> {
                // 可以在此睡眠一下
                /*try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }*/
                System.out.println("我是子线程：" + Thread.currentThread().getName());
            });
        }
        System.out.println("一共往线程池提交了10个任务");
        System.out.println("线程池是否已关闭:" + threadPool.isShutdown());
        // 发起关闭线程池
        threadPool.shutdown();
        System.out.println("线程池是否已关闭:" + threadPool.isShutdown());
        System.out.println("线程池是否已终止:" + threadPool.isTerminated());
        threadPool.awaitTermination(1, TimeUnit.SECONDS);
        if (threadPool.isTerminated()) {
            System.out.println("线程池是否已终止:" + threadPool.isTerminated());
        } else {
            System.out.println("线程池未终止（超时）就退出");
        }
    }
}

class CustomThreadFactory implements ThreadFactory {
    // 原子类 CAS 保证线程安全
    private AtomicInteger threadCount = new AtomicInteger();

    private boolean isDaemon;
    private int priority;

    public CustomThreadFactory(boolean isDaemon, int priority) {
        this.isDaemon = isDaemon;
        this.priority = priority;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r);
        thread.setName("My_Custom_Thread_" + threadCount.incrementAndGet());
        thread.setDaemon(isDaemon);
        thread.setPriority(priority);
        return thread;
    }
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61

    执行结果：

    Connected to the target VM, address: '127.0.0.1:53514', transport: 'socket'
开始往线程池中提交任务
一共往线程池提交了10个任务
线程池是否已关闭:false
线程池是否已关闭:true
线程池是否已终止:false
我是子线程：My_Custom_Thread_1
我是子线程：My_Custom_Thread_1
我是子线程：My_Custom_Thread_1
我是子线程：My_Custom_Thread_2
我是子线程：My_Custom_Thread_3
我是子线程：My_Custom_Thread_4
我是子线程：My_Custom_Thread_5
我是子线程：My_Custom_Thread_6
我是子线程：My_Custom_Thread_7
我是子线程：My_Custom_Thread_8
线程池是否已终止:true
Disconnected from the target VM, address: '127.0.0.1:53514', transport: 'socket'
Process finished with exit code 0
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    5.7 Handler

      通过这个参数你可以自定义任务的拒绝策略。如果线程池中所有的线程都在忙碌，并且工作队列也满了（前提是工作队列是有界队列），那么此时提交任务，线程池就会拒绝接收。

      至于拒绝的策略，你可以通过 handler 这个参数来指定。

    ThreadPoolExecutor 已经提供了以下 4 种策略。

    • CallerRunsPolicy：提交任务的线程自己去执行该任务。
    • AbortPolicy：默认的拒绝策略，会 throws RejectedExecutionException。
    • DiscardPolicy：直接丢弃任务，没有任何异常抛出。
    • DiscardOldestPolicy：丢弃最老的任务，其实就是把最早进入工作队列的任务丢弃，然后把新任务加入到工作队列。

      注意：使用有界队列，当任务过多时，线程池会触发执行拒绝策略，线程池默认的拒绝策略会 throw RejectedExecutionException 这是个运行时异常，对于运行时异常编译器并不强制 catch 它，所以开发人员很容易忽略。因此默认拒绝策略要慎重使用。如果线程池处理的任务非常重要，建议自定义自己的拒绝策略；并且在实际工作中，自定义的拒绝策略往往和降级策略配合使用。下面我给出一个完整的示例代码：

    public static class CallerRunsEnhancedPolicy implements RejectedExecutionHandler {
    public CallerRunsEnhancedPolicy() {}

    /**
     * Executes task r in the caller's thread, unless the executor has been shut down, in which case the task is
     * discarded.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 可以在此加一条日志
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

public static class AbortEnhancedPolicy implements RejectedExecutionHandler {
    public AbortEnhancedPolicy() {}

    /**
     * Always throws RejectedExecutionException.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     * @throws RejectedExecutionException always.
     */
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 可以在此加一条日志
        throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString());
    }
}

public static class RetryPolicy implements RejectedExecutionHandler {
    public RetryPolicy() {}

    /**
     * 提醒并进入后补线程池排队执行
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 可以在此加一条日志
        if (!e.isShutdown()) {
            // 可以在此加一条日志
            try {
                // ThreadPool.getAlternateExecutor()返回的是一个线程池，这个线程池应该是个全局的线程池，最好只初始化一次，保证其线程的安全性（可以用单例模式来初始化）
                ThreadPool.getAlternateExecutor().execute(r);
            } catch (Exception ex) {
                // 可以在此加一条日志
                throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString()
                        + "; failure retry execute task in " + ThreadPool.getAlternateExecutor().toString());
            }
        }
    }
}

public static class WaitPolicy implements RejectedExecutionHandler {
    public WaitPolicy() {}

    /**
     * 提醒并等待，直到加入线程池队列中
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 可以在此加一条日志
        if (!e.isShutdown()) {
            // 可以在此加一条日志
            try {
                e.getQueue().put(r);
                // 可以在此加一条日志
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
                // 可以在此加一条日志
                throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString()
                        + "; waiting put task to pool interrupted");
            } catch (Exception ex) {
                // 可以在此加一条日志
                throw new RejectedExecutionException(
                        "Task " + r.toString() + " rejected from " + e.toString() + "; failure put task to pool");
            }
        }
    }
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92

    下面是ThreadPool的定义：

    public final class ThreadPool {
public static final int LOGICAL_CPU_CORE_COUNT = Runtime.getRuntime().availableProcessors();
public static final int MAX_POOLSIZE_THRESHOLD = 2000;

private static Object alternateExecutorLock = new Object();
private static volatile ListenableThreadPoolExecutor alternateExecutor = null;

public static ThreadPoolExecutor getAlternateExecutor() {
    synchronized (alternateExecutorLock) {
        if (alternateExecutor == null) {
            initAlternateExecutor();
        } else if (alternateExecutor.isShutdown()) {
            boolean term = false;
            while (!term) {
                try {
                    term = alternateExecutor.awaitTermination(1, TimeUnit.SECONDS);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
            initAlternateExecutor();
        }
        return alternateExecutor;
    }
}

private static void initAlternateExecutor() {
    int threadCount =
            getRecommendedThreadPoolSizeByTaskIntensiveType(TaskIntensiveType.CPU, MAX_POOLSIZE_THRESHOLD);
    int maxThreadCount = getMaxThreadPoolSizeByTaskIntensiveType(TaskIntensiveType.CPU, MAX_POOLSIZE_THRESHOLD);
    alternateExecutor = newThreadPool(threadCount, maxThreadCount, 60L, maxThreadCount * 3,
            new WaitPolicy(), (new ThreadFactoryBuilder()).setNameFormat("alternate-pool-thread-%d").build());
    alternateExecutor.allowCoreThreadTimeOut(true);
}

public static int getRecommendedThreadPoolSizeByTaskIntensiveType(TaskIntensiveType taskIntensive,
        int poolSizeThreshold) {
    int poolSize;
    switch (taskIntensive) {
        case IO:
            poolSize = 2 * LOGICAL_CPU_CORE_COUNT + 1; // NOSONAR
            break;
        case CPU:
        default:
            poolSize = LOGICAL_CPU_CORE_COUNT + 1;
            break;
    }
    if (poolSize > poolSizeThreshold) {
        return poolSizeThreshold;
    }
    return poolSize;
}

public static int getMaxThreadPoolSizeByTaskIntensiveType(TaskIntensiveType taskIntensive, int poolSizeThreshold) {
    int poolSize;
    switch (taskIntensive) {
        case IO:
            poolSize = 2 * LOGICAL_CPU_CORE_COUNT * 5 + 1; // NOSONAR
            break;
        case CPU:
        default:
            poolSize = LOGICAL_CPU_CORE_COUNT * 5 + 1;
            break;
    }
    if (poolSize > poolSizeThreshold) {
        return poolSizeThreshold;
    }
    return poolSize;
}

public enum TaskIntensiveType {
    IO(0), CPU(1);

    private int value;

    private TaskIntensiveType(int value) {
        this.value = value;
    }

    public static TaskIntensiveType valueOf(int value) {
        for (TaskIntensiveType oneVal : TaskIntensiveType.values()) {
            if (oneVal.value == value) {
                return oneVal;
            }
        }
        return null;
    }

    public int getValue() {
        return value;
    }

}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93


    6、ThreadPoolExecutor常见方法

    6.1 execute（）

      方法实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行。

    6.2 submit（）

      方法是在ExecutorService中声明的方法，在AbstractExecutorService就已经有了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()方法的实现，会发现它实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果。submit可以提交Runnable和 Callable。

    6.3 shutdown()

      将线程池状态置为SHUTDOWN。平滑的关闭ExecutorService，当此方法被调用时，ExecutorService停止接收新的任务并且等待已经提交的任务（包含提交正在执行和提交未执行）执行完成。当所有提交任务执行完毕，线程池即被关闭。所以手动调用shotdown方法，可以不必担心存在剩余任务没有执行的情况。

    6.4 shutdownNow()

      将线程池状态置为STOP。跟shutdown()一样，先停止接收外部提交的任务，忽略队列里等待的任务，尝试将正在跑的任务interrupt中断，返回未执行的任务列表。

      对于那些正在执行的task，并不能保证他们就一定会直接停止执行，或许他们会暂停，或许会执行直到完成，但是ExecutorService会尽力关闭所有正在运行的task。

    6.5 awaitTermination(long timeout, TimeUnit unit)

      awaitTermination方法接收timeout和TimeUnit两个参数，用于设定超时时间及单位。当等待超过设定时间时，会监测ExecutorService是否已经关闭，若关闭则返回true，否则返回false。一般情况下会和shutdown方法组合使用。

      第一个参数指定的是时间，第二个参数指定的是时间单位(当前是秒)。返回值类型为boolean型。

      如果等待的时间超过指定的时间，但是线程池中的线程运行完毕，那么awaitTermination()返回true。执行分线程已结束。

      如果等待的时间超过指定的时间，但是线程池中的线程未运行完毕，那么awaitTermination()返回false。不执行分线程已结束。
    如果等待时间没有超过指定时间，等待！

      可以用awaitTermination()方法来判断线程池中是否有继续运行的线程。
    下面是ThreadPoolExecutor的awaitTermination源码：

    /**
 * Wait condition to support awaitTermination
 */
private final Condition termination = mainLock.newCondition();

public boolean awaitTermination(long timeout, TimeUnit unit)
    throws InterruptedException {
// 等待时长
    long nanos = unit.toNanos(timeout);
//加锁
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (;;) {
            if (runStateAtLeast(ctl.get(), TERMINATED))
                return true;
            if (nanos <= 0)
                return false;
	// condition条件等待
            nanos = termination.awaitNanos(nanos);
        }
    } finally {
        mainLock.unlock();
    }
}
     
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25

    6.6 其它方法

    isShutdown()：线程池是否已关闭

    isTerminated：线程池是否已终止



    7、线程池是否需要关闭呢？

      这里说下两种情况下的线程池：全局线程池、局部线程池。

    • 全局线程池：其实我们工作中使用线程池都是全局的，当然我们也可能为不同业务建立不同的全局线程池。那全局线程池是否需要关闭呢？分两种场景：（一）如果每次在程序的结尾都去关闭线程池，那每次有新的任务进来，都要重新建立一个线程池，这样难免也太消耗资源了，而且原本这就是一个全局的。（二）如果一个线程池在某段时间内处理了大量的任务，创建了大量的线程，这个时间段之后可能一下子也没有新的任务进来，那这个线程池需要关闭嘛 ？我个人理解也是没必要的，我们通过参数keepAliveTime可以给线程设置在没有任务处理时的生存时间，并且调用allowCoreThreadTimeOut(boolean)方法，这样充分减少资源的浪费。

    • 局部线程池：这种线程池工作中也该尽量避免使用。如果每个请求进来都创建一个新的线程池，当请求多的时候，线程池增加，那难免会有OOM。如果你真的使用了，那请记住一定要调用shutdown()来关闭。

    推荐阅读：
    Java内存模型-volatile的应用（实例讲解）
    大彻大悟synchronized原理，锁的升级
  • 相关阅读:
    JS开发框架DevExtreme v20.1.7上线
    Web开发：看如何定义Kendo UI Grid Width
    如何创建自定义DevExpress报表控件，看完你就懂了
    高性能HTML5/JavaScript开发框架DevExtreme全新发布v20.1.7|附下载
    WPF界面开发：如何将不同集合中的项目显示为同一父节点子项
    界面开发包DevExpress v20.1.7上线！即刻体验
    WPF界面开发2020：Scheduler等控件功能升级
    Winform界面开发看过来！一招教你使用属性网格自定义编辑器
    将从数据库查询出来的带有父子结构的list转换成treeList结构
    将JDBC ResultSet结果集转成List
  • 原文地址：https://www.cnblogs.com/cdchencw/p/15137244.html
Copyright © 2011-2022 走看看