进程、线程与线程池

吹梦到西洲
多线程与并发

发布于：Jul 19, 2020
次浏览

进程

什么是进程

  • 狭义定义：进程是正在运行的程序的实例。
  • 广义定义：进程是一个具有一定独立功能的程序关于某个数据集合的一次运行活动。它是操作系统动态执行的基本单元，在传统的操作系统中，进程既是基本的分配单元，也是基本的执行单元。
  • 进程的概念主要有两点：
    • 进程是一个实体(程序段、相关的数据段、PCB)：每一个进程都有它自己的地址空间，一般情况下，包括文本区域 $text\ region$、数据区域 $data\ region$ 和堆栈 $stack\ region$。文本区域存储处理器执行的代码；数据区域存储变量和进程执行期间使用的动态分配的内存；堆栈区域存储着活动过程调用的指令和本地变量。
    • 进程是一个“执行中的程序”：程序是一个没有生命的实体，只有处理器赋予程序生命时(操作系统执行之)，它才能成为一个活动的实体，我们称其为进程。进程由创建而产生，由调度而执行，由撤销而消亡。

《计算机操作系统》第三版对进程的定义是：

  • 进程是程序的一次执行。
  • 进程是一个程序及其数据在处理机上顺序执行时所发生的活动。
  • 进程是程序在一个数据集合上运行的过程，它是系统进行资源分配和调度的一个独立单位。

进程的状态

三态模型

  • 就绪($Ready$)状态

当进程已分配到除 $CPU$ 以外的所有必要资源后，只要再获得 $CPU$，便可立即执行，进程这时的状态称为就绪状态。在一个系统中处于就绪状态的进程可能有多个，通常将它们排成一个队列，称为就绪队列。

  • 执行状态

进程已获得 $CPU$，其程序正在执行。在单处理机系统中，只有一个进程处于执行状态；在多处理机系统中，则有多个进程处于执行状态。

  • 阻塞状态

正在执行的进程由于发生某事件而暂时无法继续执行时，便放弃处理机而处于暂停状态，亦即进程的执行受到阻塞，把这种暂停状态称为阻塞状态，有时也称为等待状态或封锁状态。致使进程阻塞的典型事件有：请求$I/O$，申请缓冲空间等。通常将这种处于阻塞状态的进程也排成一个队列。有的系统则根据阻塞原因的不同而把处于阻塞状态的进程排成多个队列。

进程的三种基本状态及其转换

五态模型

在不少系统中进程只有上述三种状态，但在另一些系统中，又增加了一些新状态，最重要的是挂起状态。引入挂起状态的原因有：

  • 终端用户的请求。当终端用户在自己的程序运行期间发现有可疑问题时，希望暂时使自己的程序静止下来。
  • 父进程请求。有时父进程希望挂起自己的某个子进程，以便考查和修改该子进程，或者协调各子进程间的活动。
  • 负荷调节的需要。当实时系统中的工作负荷较重，已可能影响到对实时任务的控制时，可由系统把一些不重要的进程挂起，以保证系统能正常运行。
  • 操作系统的需要。操作系统有时希望挂起某些进程，以便检查运行中的资源使用情况或进行记账。

具有挂起状态的进程状态图

在目前实际的系统中，为了管理的需要，还存在着两种比较常见的进程状态，即创建状态和终止状态。

  • 创建状态

创建一个进程一般要通过两个步骤：首先，为一个新进程创建 $PCB$，并填写必要的管理信息；其次，把该进程转入就绪状态并插入就绪队列之中。

  • 终止状态

进程的终止也要通过两个步骤：首先等待操作系统进行善后处理，然后将其 $PCB$ 清零，并将 $PCB$ 空间返还系统。

进程的五种基本状态及转换

增加了创建状态和终止状态后，具有挂起状态的进程状态及转换图。

具有创建、终止和挂起状态的进程状态图

进程同步方式

  • 信号量机制
    • 整型信号量
    • 记录型信号量
    • $AND$ 型信号量
    • 信号量集
  • 管程机制

管程由四部分组成：① 管程的名称；② 局部于管程内部的共享数据结构说明；③ 对该数据结构进行操作的一组过程；④ 对局部于管程内部的共享数据设置初始值的语句。

管程的示意图

进程通信方式

  1. 管道/匿名管道(Pipes) ：用于具有亲缘关系的父子进程间或者兄弟进程之间的通信。
  2. 有名管道(Names Pipes) : 匿名管道由于没有名字，只能用于亲缘关系的进程间通信。为了克服这个缺点，提出了有名管道。有名管道严格遵循先进先出(first in first out)。有名管道以磁盘文件的方式存在，可以实现本机任意两个进程通信。
  3. 信号(Signal) ：信号是一种比较复杂的通信方式，用于通知接收进程某个事件已经发生；
  4. 消息队列(Message Queuing) ：消息队列是消息的链表,具有特定的格式,存放在内存中并由消息队列标识符标识。管道和消息队列的通信数据都是先进先出的原则。与管道（无名管道：只存在于内存中的文件；命名管道：存在于实际的磁盘介质或者文件系统）不同的是消息队列存放在内核中，只有在内核重启(即，操作系统重启)或者显示地删除一个消息队列时，该消息队列才会被真正的删除。消息队列可以实现消息的随机查询,消息不一定要以先进先出的次序读取,也可以按消息的类型读取.比 FIFO 更有优势。消息队列克服了信号承载信息量少，管道只能承载无格式字 节流以及缓冲区大小受限等缺。
  5. 信号量(Semaphores) ：信号量是一个计数器，用于多进程对共享数据的访问，信号量的意图在于进程间同步。这种通信方式主要用于解决与同步相关的问题并避免竞争条件。
  6. 共享内存(Shared memory) ：使得多个进程可以访问同一块内存空间，不同进程可以及时看到对方进程中对共享内存中数据的更新。这种方式需要依靠某种同步操作，如互斥锁和信号量等。可以说这是最有用的进程间通信方式。
  7. 套接字(Sockets) : 此方法主要用于在客户端和服务器之间通过网络进行通信。套接字是支持 TCP/IP 的网络通信的基本操作单元，可以看做是不同主机之间的进程进行双向通信的端点，简单的说就是通信的两方的一种约定，用套接字中的相关函数来完成通信过程。

进程调度算法

  • 先到先服务(FCFS)调度算法 : 从就绪队列中选择一个最先进入该队列的进程为之分配资源，使它立即执行并一直执行到完成或发生某事件而被阻塞放弃占用 CPU 时再重新调度。
  • 短作业优先(SJF)的调度算法 : 从就绪队列中选出一个估计运行时间最短的进程为之分配资源，使它立即执行并一直执行到完成或发生某事件而被阻塞放弃占用 CPU 时再重新调度。
  • 时间片轮转调度算法 : 时间片轮转调度是一种最古老，最简单，最公平且使用最广的算法，又称 RR(Round robin)调度。每个进程被分配一个时间段，称作它的时间片，即该进程允许运行的时间。
  • 多级反馈队列调度算法 ：前面介绍的几种进程调度的算法都有一定的局限性。如短进程优先的调度算法，仅照顾了短进程而忽略了长进程 。多级反馈队列调度算法既能使高优先级的作业得到响应又能使短作业（进程）迅速完成。，因而它是目前被公认的一种较好的进程调度算法，UNIX 操作系统采取的便是这种调度算法。
  • 优先级调度 ： 为每个流程分配优先级，首先执行具有最高优先级的进程，依此类推。具有相同优先级的进程以 FCFS 方式执行。可以根据内存要求，时间要求或任何其他资源要求来确定优先级。

线程

进程与线程的区别

进程是一个独立的运行环境，而线程是在进程中执行的一个任务。他们两个本质的区别是是否单独占有内存地址空间及其它系统资源（比如I/O）：

  • 进程单独占有一定的内存地址空间，所以进程间存在内存隔离，数据是分开的，数据共享复杂但是同步简单，各个进程之间互不干扰；而线程共享所属进程占有的内存地址空间和资源，数据共享简单，但是同步复杂。
  • 进程单独占有一定的内存地址空间，一个进程出现问题不会影响其他进程，不影响主程序的稳定性，可靠性高；一个线程崩溃可能影响整个程序的稳定性，可靠性较低。
  • 进程单独占有一定的内存地址空间，进程的创建和销毁不仅需要保存寄存器和栈信息，还需要资源的分配回收以及页调度，开销较大；线程只需要保存寄存器和栈信息，开销较小。

另外一个重要区别是，进程是操作系统进行资源分配的基本单位，而线程是操作系统进行调度的基本单位，即 $CPU$ 分配时间的单位 。

《深入理解Java虚拟机》第三版中提到，一种名为纤程的轻量级线程正在逐渐成熟。

进程与线程的关系

  • $Java$ 对操作系统提供的功能进行封装，包括进程与线程。
  • 运行一个程序会产生一个进程，一个进程至少包含一个线程。
  • 每个进程对应一个 $JVM$ 实例，多个线程共享 $JVM$ 里的堆。
  • $Java$ 采用单线程编程模型，程序会自动创建主线程。
  • 主线程可以创建子线程，原则上要晚于子线程完成执行。

线程同步的方式

为使系统中的多线程能有条不紊地运行，在系统中必须提供用于实现线程间同步和通信的机制。为了支持不同频率的交互操作和不同程度的并行性，在多线程 $OS$ 中通常提供多种同步机制，如互斥锁、条件变量、计数信号量以及多读、单写锁等。

  1. 互斥锁($mutex$)

互斥锁可以有开锁(unlock)和关锁(lock)两种状态。相应地，可用两条命令(函数)对互斥锁进行操作。其中的关锁lock操作用于将mutex关上，开锁操作unlock则用于打开mutex。

当一个线程需要读/写一个共享数据段时，线程首先应为该数据段所设置的mutex执行关锁命令。

  • 命令首先判别mutex的状态，如果它已处于关锁状态，则试图访问该数据段的线程将被阻塞；而如果mutex是处于开锁状态，则将mutex关上后便去读/写该数据段。
  • 在线程完成对数据的读/写后，必须再发出开锁命令将mutex打开，同时还须将阻塞在该互斥锁上的一个线程唤醒，其它的线程仍被阻塞在等待mutex打开的队列上。

  1. 条件变量
    每一个条件变量通常都与一个互斥锁一起使用，亦即，在创建一个互斥锁时便联系着一个条件变量。单纯的互斥锁用于短期锁定，主要是用来保证对临界区的互斥进入。而条件变量则用于线程的长期等待，直至所等待的资源成为可用的资源。

  2. 信号量机制

    就是信号量机制。

线程创建

  • 继承Thread类
    • 重写Thread::run；
    • 调用Thread::start。
  • 实现Runnable接口，无返回值。
    • 获取Runnable接口的实现类，作为参数，创建Thread；
    • 执行Thread::start。
  • 实现Callable接口，结合FutureTask使用，有返回值。
    • 以Callable的实现类为参数，创建FutureTask；
    • 将FutureTask作为Thread的参数，创建Thread;
    • 通过Thread::start启动线程；
    • 通过FutureTask::get阻塞获取线程返回值。
  • 利用线程池Executor
    • 创建Callable或Runnable任务，提交到线程池；
    • 通过返回的Future::get获取返回结果。

线程状态

线程状态

  • $NEW$：新建，创建后尚未启动。
  • $RUNNABLE$：运行，包含 $Running$ 和 $Ready$
  • $BLOCKED$：阻塞，等待获取排它锁
  • $WAITING$ ：等待，不会被分配 CPU 时间，需要显示被唤醒
  • $TIMED-WAITING$ ：限期等待，在一定时间后会由系统自动唤醒
  • $TERMINATED$： 终止，已终止线程的状态，线程已执行结束。

线程状态图

start() 与 run()

  • start() : 通过start()方法来启动的新线程，处于就绪状态，并没有运行，一旦得到 $CPU$ 时间片，就开始执行相应线程的run()方法，这里方法run()称为线程体，它包含了要执行的这个线程的内容，run()方法运行结束，此线程随即终止。start()不能被重复调用。用start()方法来启动线程，真正实现了多线程运行，即无需等待某个线程的run()方法体代码执行完毕就直接继续执行下面的代码，即进行了线程切换。
  • run() : run() 就和普通的成员方法一样，可以被重复调用。如果直接调用run()方法，并不会启动新线程！程序中依然只有主线程这一个线程，其程序执行路径还是只有一条，还是要顺序执行，还是要等待run()方法体执行完毕后才可继续执行下面的代码，这样就没有达到多线程的目的。
  • 总结：调用 start() 方法会创建一个新的子线程并启动， run() 方法只是 Thread 的一个方法调用，还是在原线程里运行。

如何给run()方法传参

构造函数传参、成员变量传参、回调函数传参。

sleep() 与 wait()

  • sleep() 是 Thread 类的方法， wait() 是 Object 类的方法。
  • sleep() 方法可以在任何地方使用；wait() 方法只能在 synchronized 方法或 synchronized 块中使用。
  • Thread.sleep() : 只会让出$CPU$，不会导致锁行为的改变；Object.wait() : 不仅让出 $CPU$，还会释放已经占有的同步资源锁。

notify() 与 notifyAll()

  • 锁池($Entry\ List$)：假设线程 $A$ 已经拥有了某个对象(注意:不是类)的锁，而其它的线程想要调用这个对象的某个 synchronized 方法(或者 synchronized块)，由于这些线程在进入对象的 synchronized 方法之前必须先获得该对象的锁的拥有权，但是该对象的锁目前正被线程 $A$ 拥有，所以这些线程就进入了该对象的锁池中。

  • 等待池($Wait\ Set$): 假设一个线程 $A$ 调用了某个对象的 wait() 方法，线程 $A$ 就会释放该对象的锁(因为 wait() 方法必须出现在 synchronized 中，这样自然在执行 wait() 方法之前线程 $A$ 就已经拥有了该对象的锁)，同时线程 $A$ 就进入到了该对象的等待池中，进入到等待池中的线程不会去竞争该对象的锁。如果另外的一个线程调用了相同对象的 notifyAll() 方法，那么处于该对象的等待池中的线程就会全部进入该对象的锁池中，准备争夺锁的拥有权。如果另外的一个线程调用了相同对象的 notify() 方法，那么仅仅有一个处于该对象的等待池中的线程(随机)会进入该对象的锁池。

  • notify() : 只会随机选取一个处于等待池中的线程进入锁池去竞争获取锁的机会。

  • notifyAll() : 会让所有处于等待池中的线程全部进入锁池去竞争获取锁的机会。

yield()

当调用 Thread.yield() 方法时，会给线程调度器一个当前线程愿意让出 $CPU$ 使用的暗示，但是线程调度器可能会忽略这个暗示。

如何实现处理线程的返回值

  • 主线程等待
  • 使用 Thread::join 方法阻塞当前线程以等待子线程处理完毕
  • 通过 Callable 接口实现：通过 FutureTask<> 或者 Executor 获取

如何中断线程

  • 调用 interrupt() ，通知线程应该中断了
    • 如果线程处于被阻塞状态，那么线程将立即退出被阻塞状态，并抛出一个 InterruptedException 异常。
    • 如果线程处于正常活动状态，那么会将该线程的中断标志设置为 true ，被设置中断标志的线程将继续正常运行，不受影响。

线程池

为什么要使用线程池

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

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

线程池的使用其实非常广泛，比如MySQL和Tomcat为客户端连接准备的线程池。

线程池集合

Executor 框架

上图中有三个重要的 Executor 接口

  • Executor：运行新任务的简单接口，将任务提交和任务执行细节解耦。
  • ExecutorService：具备管理执行器和任务生命周期的方法，提交任务机制更完善。
  • ScheduledExecutorService：支持 Future 和定期执行任务

线程池参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    // corePoolSize: 核心线程数量
    // maximumPoolSize: 线程不够用时能够创建的最大线程数
    // keepAliveTime: 存活等待时间
    // unit: 时间单位
    // workQueue: 任务等待队列
    // threadFactory: 默认 Executors.defaultThreadFactory()
    // handler: 线程池饱和时的拒绝策略
    
    // ...
}

阻塞队列

BlockingQueue workQueue：阻塞队列，维护着等待执行的Runnable任务对象。

常用的几个阻塞队列：

  • LinkedBlockingQueue：链式阻塞队列，底层数据结构是链表，默认大小是Integer.MAX_VALUE，也可以指定大小。

  • ArrayBlockingQueue：数组阻塞队列，底层数据结构是数组，需要指定队列的大小。

  • SynchronousQueue：同步队列，内部容量为0，每个put操作必须等待一个take操作，反之亦然。

  • DelayQueue：延迟队列，该队列中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素 。

线程工厂

创建线程的工厂 ，用于批量创建线程，统一在创建线程时设置一些参数，如是否守护线程、线程的优先级等。如果不指定，会新建一个默认的线程工厂。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
static class DefaultThreadFactory implements ThreadFactory {
    // ...
    
    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
        Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
            poolNumber.getAndIncrement() +
            "-thread-";
    }

    // ...
}

饱和策略

线程池饱和时的拒绝策略

  • AbortPolicy : 直接抛出异常，这是默认策略
  • CallerRunsPolicy : 用调用者所在的线程来执行任务
  • DiscardOldestPolicy : 丢弃队列中最靠前的任务，并执行当前任务
  • DiscardPolicy : 直接丢弃任务
  • 实现 RejectedExecutionHandler 接口的自定义 Handler

线程池工作流程

处理任务的核心方法是execute()

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
// JDK 1.8 
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();   
    int c = ctl.get();
    // 1.当前线程数小于 corePoolSize, 则调用 addWorker 创建核心线程执行任务.
    if (workerCountOf(c) < corePoolSize) {
       if (addWorker(command, true))
           return;
       c = ctl.get();
    }
    // 2.如果不小于 corePoolSize, 则将任务添加到 workQueue 队列.
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 2.1 如果 isRunning 返回 false (状态检查), 则 remove 这个任务, 然后执行拒绝策略. 
        if (!isRunning(recheck) && remove(command))
            reject(command);
            // 2.2 线程池处于 running 状态, 但是没有线程, 则创建线程.
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 3.如果放入 workQueue 失败, 则创建非核心线程执行任务.
    // 如果这时创建非核心线程失败(当前线程总数不小于 maximumPoolSize 时), 就会执行拒绝策略.
    else if (!addWorker(command, false))
         reject(command);
}

新任务提交 execute 执行后的判断

  • 如果运行的线程少于 corePoolSize ，则创建新线程来处理任务，即使线程池中的其它线程是空闲的。
  • 当 workQueue 未满，线程池中的线程数量大于等于 corePoolSize 且小于 maximumPoolSize，将任务加到workQueue中进行等待，等待空闲coreThread来处理任务。
  • 当 workQueue 已满，并且正在运行的线程数量大于等于corePoolSize并且小于maximumPoolSize，才创建新的线程去处理任务。
  • 如果运行的线程数量大于等于 maximumPoolSize ，这时如果 workQueue 已经满了，则通过 Handler 所制定的策略来处理任务。

线程池执行流程图

img

线程复用

当一个线程被创建的时候会被指定一个任务，当执行完这个任务之后线程会自动销毁。但是线程池可以复用线程，也就是线程在执行完任务后不被销毁，继续执行其他的任务。那么，线程池是如何做到线程复用的呢？

其实，ThreadPoolExecutor在创建线程时，会将线程封装成工作线程worker，并放入工作线程组中，然后这个工作线程worker反复从阻塞队列中取任务执行。

瞥一眼工作线程Worker

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
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    
    // 这个类不会被序列化, 提供了序列号只是为了祛除 javac 的警告.
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** 每个线程完成的任务数量 */
    volatile long completedTasks;

    // 根据从线程工厂获得的 first task 和 thread 创建
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }

    // 重写继承自 AQS 的方法
    // 0 代表解锁状态 1 代表加锁状态

    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }

    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

线程池的状态

  • RUNNING : 线程池刚创建后的状态，能接受新提交的任务，并且也能处理阻塞队列中的任务。
  • SHUTDOWN : 调用shutdown()方法后，不再接受新提交的任务，但可以处理存量任务。
  • STOP : 调用shutdownNow()方法后，不再接受新提交的任务，也不再处理存量任务。中断所有线程，阻塞队列中没有执行完的任务全部丢弃。
  • TIDYING : 所有的任务都已终止，ctl记录的任务数量为 $0$，接下来会执行terminated()进入$TERMINATED$ 状态。
  • TERMINATED : terminated() 方法执行完后进入该状态。
    线程池状态转换图

线程池的大小如何选定

  • $CPU$ 密集型：线程数 $=$ 核数或者核数 $ + \ 1$
  • $IO$ 密集型：线程数 $=$ $CPU$核数 $\times$ ($1 +$ 平均等待时间/平均工作时间)

常见的线程池

Executors类提供了几个静态方法用来创建线程池，但是在实际项目中并不推荐直接使用，而是采用手动创建的方式。这样能够提醒创建者更加明确实际任务，避免资源耗尽的风险。

newSingleThreadExecutor()

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

有且仅有一个核心线程(corePoolSize == maximumPoolSize == 1)，使用了LinkedBlockingQueue(容量很大)，所以，不会创建非核心线程，所有任务按照先来先执行的顺序执行。如果这个唯一的线程不空闲，那么新来的任务会存储在任务队列里等待执行。

newCachedThreadPool()

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

运行流程：

  1. 将任务提交到线程池；
  2. 因为corePoolSize = 0，不必创建核心线程，maximumPoolSize为Integer.MAX_VALUE，即2147483647；
  3. 尝试将任务添加到SynchronousQueue队列；
  4. 如果入队成功，等待被空闲线程拉取并执行，如果没有空闲线程，那么创建一个非核心线程，然后从队列中拉去任务并执行；
  5. 如果队列中已有任务在等待，入队操作将被阻塞。

当需要执行很多短时间的任务时，CacheThreadPool的线程复用率比较高， 会显著地提高性能。而且线程 $60s$ 后会回收，意味着即使没有任务进来，CacheThreadPool并不会占用很多资源。

newFixedThreadPool()

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

corePoolSize == maximumPoolSize，都为传入参数nThreads。所以只能创建核心线程，不能创建非核心线程。因为LinkedBlockingQueue的默认大小为Integer.MAX_VALUE。所以如果核心线程空闲，则交给核心线程处理，否则入队等待。等有核心线程空闲时进行处理。

与CachedThreadPool的区别：

  • corePoolSize == maximumPoolSize，只创建核心线程；而CachedThreadPool的corePoolSize == 0，所以只会创建非核心线程。
  • getTask()方法获取任务时：线程会阻塞在LinkedBlockingQueue.take()，线程不会被回收；CachedThreadPool会在 $60s$ 后被回收。
  • 由于线程不会被回收，会一直卡在阻塞，所以没有任务的情况下，FixedThreadPool占用资源更多。
  • 都几乎不会触发拒绝策略，但是原理不同。FixedThreadPool是因为阻塞队列可以很大，故几乎不会触发拒绝策略；CachedThreadPool是因为线程池很大，几乎不会导致线程数量大于最大线程数，故几乎不会触发拒绝策略。

newScheduledThreadPool()

创建一个定长线程池，支持定时及周期性任务执行。

1
2
3
4
5
6
7
8
9
10
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

// ScheduledThreadPoolExecutor():
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue());
}

newWorkStealingPool()

1
2
3
4
5
6
public static ExecutorService newWorkStealingPool(int parallelism) {
    return new ForkJoinPool
        (parallelism,
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}
  • 内部会构建 ForkJoinPool ，利用 work-stealing 算法，并行地处理任务，不保证处理顺序。
  • Fork/Join 框架：把大任务分割成若干个小任务并行执行，最终汇总每个小任务结果后得到大任务结果的框架。

扩展阅读：

博客内容遵循 署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0) 协议

本文永久链接是：http://raymond-zhao.top/2020/07/19/2020-07-19-JUC-ProcessAndThread/

更新于：Jun 20, 2021

线程与线程池

评论