并发基础 - 1

参考文献

• https://javabetter.cn/sidebar/sanfene/javathread.html
• https://javaguide.cn/java/concurrent/java-concurrent-questions-01.html
• https://tech.meituan.com/2020/04/02/java-pooling-pratice-in-meituan.html
• https://www.zhihu.com/question/341005993/answer/128919891202

概念辨析

并行与并发

并行与并发的区别在于，并行是同一时刻真正同时执行多个任务，并发是在一段时间内交替处理多个任务、看起来像是同时运行。

在单核CPU上的多线程程序与js在浏览器中的异步操作为并发但不并行的操作，多核CPU上的Web服务器与分布式系统上的微服务即为既并发又并行的场景。

异步与并发

异步和并发的区别在于，异步在意“等待”，允许任务在等待时不阻碍主线程；并发在乎“同时发生”，指多个任务在重叠的时间段内开始、运行和完成，但不一定在同一瞬间内完成。

异步和同步的区别在于，在发出调用后，异步返回结果前返回调用，而同步在没有得到结果前一直不反回调用。

同步, 异步, 阻塞, 非阻塞

美团后台开发 - 暑期实习, 一面.

这里实际上是两个维度: 同步和异步, 阻塞和非阻塞.

维度	含义	关键问题
同步/异步	调用方如何得知结果	同步时, 调用方主动等结果 异步时, 调用方等回应方通知
阻塞/非阻塞	在等待结果时, 当前线程能不能干其他事	阻塞时, 线程会挂起等待 非阻塞时, 线程能立刻返回, 干点别的

以上两种维度, 可以派生出四种组合:

1. 同步阻塞
   最传统的方式, 调用方发起请求, 然后一直等, 线程挂起, 知道结果返回.
2. 同步非阻塞
   调用方发起请求. 如果结果没准备好, 则立刻返回. 线程不挂起. 在一定时间过后, 会主动再问一次.
   例子: 在一个 spring task 中每隔一段时间轮询所有数据, 看里面有数据准备好了没.
3. 异步阻塞
   一般少见. 调用方发起异步操作后, 立刻去等那个结果. 线程挂起直到结果返回.
4. 异步非阻塞
   最理想的模型. 调用方发起请求后立即返回, 不等待结果. 当结果准备好时, 系统通过回调, 事件, 信号等方式通知调用方. 整个过程中调用方线程从未挂起.
   例子: 一般大模型, ASR 等处理都需要时间, 因此在一些 SaaS 服务中可以提供回调服务.

概述

JDK 1.2 之前，Java 线程就是用户线程。JDK 1.2 后，Java 采用一对一内核线程映射模型，每创建一个 Thread 对象并调用 start() 方法时，JVM 会通过操作系统原生 API 向内核申请创建对应的原生操作系统线程，Java 线程的生命周期、调度完全交由操作系统内核调度器管理。

Java 的线程和普通线程一样，也需要存储一些私有数据，程序计数器用于线程切换后能恢复到正确是执行位置，虚拟机栈用于存储局部变量表和常量池引用等信息，本地方法栈用于存储虚拟机使用到的 native 方法。

创建

线程的创建有三种方式。

第一种是重写父类Thread的run()方法，并且调用start()方法启动线程。

class ThreadTask extends Thread {
    public void run() {
        // ...
    }

    public static void main(String[] args) {
        ThreadTask task = new ThreadTask();
        task.start();
    }
}

该方法的缺点是如果类已经继承了另外一个类，就不能再继承Thread类了。

第二种是实现Runnable接口的run()方法。并将实现类的对象作为参数传递给Thread对象的构造方法，最后调用start()方法启动线程。

class RunnableTask implements Runnable {
    public void run() {
        // ...
    }

    public static void main(String[] args) {
        RunnableTask task = new RunnableTask();
        Thread thread = new Thread(task);
        thread.start();
    }
}

第三种需要重写Callable接口的call()方法，然后创建FutureTask对象，参数为Callable实现类的对象，紧接着创建Thread对象，参数为FutureTask对象，然后调用start()方法启动线程。

class CallableTask implements Callable<String> {
    public String call() {
        // ...
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CallableTask task = new CallableTask();
        FutureTask<String> futureTask = new FutureTask<>(task);
        Thread thread = new Thread(futureTask);
        thread.start();
        System.out.println(futureTask.get());
    }
}

该方法能获取线程的执行结果。

创建线程的时候，至少需要分配一个虚拟机栈，在64位操作系统中，默认大小为1M。当然，由于JVM、操作系统本身运行也需要内存，因此只给这么大的内存卡跑到理论值肯定是不够的。

Runnable 和 Callable 的区别如下

特性	Runnable	Callable
包路径	java.lang.Runnable	java.util.concurrent.Callable
方法签名	void.run()	V call() throw Exception
适用场景	简单的异步任务	需要返回结果或处理异常的任务

Runnable 在 Java 1.0 以来一直存在，但 Callable 在 Java 1.5 后才引入，就是为了处理 Runnable 不支持的用例。工具类 Executors 可以实现将 Runnable 对象转换成 Callable 对象（Executors.callable(Runnable task) 或 Executors.callable(Runnable task, Object result)）

调用start()的时候会执行run()方法，那为什么不直接调用run()方法？因为start()方法是启动一个新的线程，然后让这个线程去执行run()方法。而run()方法本身只是在当前线程中调用的一个普通方法。
调用start()后，线程进入就绪状态，等待操作系统调度；一旦调度执行，线程会执行其run()方法中的代码。

如果直接执行 run() 方法，就不会以多线程的方式去执行了。

调度

指令

1. 线程启动与任务执行
   • 启动线程：将线程从创建状态转为就绪状态，等待 CPU 调度执行，执行后会自动运行线程内定义的任务逻辑。
2. 线程休眠
   • 让线程暂停执行指定的毫秒级时长，休眠期间不会释放持有的对象锁。
3. 线程等待
   • 基于共享对象的等待：需在 synchronized 代码块中使用，让当前线程进入等待状态（可指定最长等待毫秒数），需其他线程唤醒或被中断时才会退出等待；
   • 等待目标线程完成：让当前线程暂停执行，直到目标线程执行完毕（可指定最长等待毫秒数）。
4. 线程唤醒
   • 唤醒单个等待线程：随机唤醒一个在该共享对象上等待的线程；
   • 唤醒所有等待线程：唤醒所有在该共享对象上等待的线程。
5. 线程让步与中断
   • 线程让步：提示操作系统当前线程可让出 CPU 执行权，但不保证立刻让出，线程仅从运行状态回到就绪状态；
   • 线程中断：修改线程的中断状态，不会直接终止运行中的线程，需线程自行处理中断标志；
   • 中断状态检查：仅查询线程的中断状态，不修改状态值；
   • 中断状态检查并清除：查询线程中断状态的同时，将中断状态重置为未中断。
6. 线程优先级调整
   • 设置优先级：为线程指定 1~10 的优先级数值，仅作为操作系统调度的建议，不保证线程的执行顺序；
   • 获取优先级：查询当前线程的优先级数值。
7. 守护线程管理
   • 设置守护线程：将线程标记为守护线程（参数为布尔值），当所有非守护线程执行完毕后，JVM 会直接退出，无需等待守护线程执行完成；
   • 检查守护线程：判断当前线程是否为守护线程。守护线程通常用于为其他线程提供服务。

• 线程调用共享对象的等待操作后会被挂起，直到满足以下任一条件：
  1. 其他线程调用该共享对象的唤醒操作；
  2. 其他线程中断该线程，导致抛出 InterruptedException 异常。
• 唤醒操作随机选择单个等待线程，且等待操作会自动释放对象锁（因此需在同步块中执行，且该方法定义在对象类中，而非 Thread 类）。
• 中断操作的特性：中断仅修改线程的中断状态，不会直接终止运行中的线程；若中断一个已处于中断状态或正在等待的线程，会触发 InterruptedException 异常。
• 休眠与等待的核心差异：休眠操作仅暂停线程执行，不会释放持有的对象锁；而等待操作会主动释放对象锁。

总结

1. Java 线程操作覆盖启动、休眠、等待、唤醒、让步、中断、优先级调整、守护线程管理八大核心场景；
2. 等待 / 唤醒操作依赖共享对象锁，等待会释放锁、休眠不释放锁，是核心差异点；
3. 中断仅修改状态不终止线程，守护线程为服务型线程，随非守护线程结束而退出。

状态

和OS课上的不大一样，Java中管理线程有6种状态

上下文切换

上下文是指线程在执行过程中自己的运行条件和状态，比如程序计数器，栈信息等。当线程从占用 CPU 状态退出时（终止运行除外），OS 需要保存当前线程的上下文以留待下次使用，并加载下一个将要占用的 CPU 线程上下文。这就是所谓的上下文切换。

通信

阻塞队列

这是最常用的通信方式（毕竟用上消息队列就得上 redis 或者各种 MQ 了）。可以简单实现生产者消费者模式。

BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);

new Thread(() -> {
    try {
        String data = "message";
        queue.put(data);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

new Thread(() -> {
    try {
        String data = queue.take();
        System.out.println("收到: " + data);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

信号量

依旧是经典的信号量和PV操作。控制并发访问资源的线程数量。可以用于流量控制，比如数据库连接池、网络连接池等。常用于线程池。

Semaphore semaphore = new Semaphore(5); // 最大5个许可

for (int i = 0; i < 10; i++) {
    new Thread(() -> {
        try {
            semaphore.acquire(); // 获取许可
            // 使用数据库连接
            System.out.println(Thread.currentThread().getName() + " 使用连接");
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            semaphore.release(); // 释放许可
        }
    }).start();
}

Exchanger 交换器

两个线程在同步点交换数据。一个线程调用exchange()方法时会阻塞，直到另一个线程也调用exchange()方法，然后两个线程交换数据后继续执行。

Exchanger 可以用于遗传算法，也可以用于校对工作，比如我们将纸制银行流水通过人工的方式录入到电子银行时，为了避免错误，可以录入两遍，然后通过 Exchanger 来校对两次录入的结果。

Exchanger<String> exchanger = new Exchanger<>();

new Thread(() -> {
    try {
        String data = "来自线程1的数据";
        String response = exchanger.exchange(data);
        System.out.println("线程1收到: " + response);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

new Thread(() -> {
    try {
        String data = "来自线程2的数据";
        String response = exchanger.exchange(data);
        System.out.println("线程2收到: " + response);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

多个线程可以通过volatile关键字和synchronized关键字来实现通信。volatile关键字确保变量的可见性，synchronized关键字确保线程之间的互斥访问。

volatile可以用于修饰成员变量，告知程序任何对该变量的访问均需要从共享内存中获取最新值，而不是从线程的本地缓存中获取。

synchronized关键字可以用于修饰方法或代码块，确保同一时刻只有一个线程可以访问被修饰的代码，从而避免数据竞争和不一致的问题。

CompletableFuture类提供了一种更简洁和强大的方式来处理异步编程和线程间通信。它允许线程在完成计算后将结果传递给其他线程，并支持链式调用和组合多个异步任务。

CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return "Hello";
}).thenApply(result -> {
    return result + " World";
}).thenAccept(finalResult -> {
    System.out.println(finalResult);
});

保证线程安全

Java中提供了许多方法来保证线程安全。

• 可以在代码块或方法上使用synchronized关键字。
• 使用ReentrantLock类来显式地加锁和解锁，这种锁还支持可重入、公平锁/非公平锁。
• 保证变量的可见性，可以使用volatile关键字修饰变量。
• 使用Atomic类（如AtomicInteger、AtomicLong等）
• 对于线程独立的数据，可以使用ThreadLocal类来存储和访问这些数据。
• 对于需要并发容器的地方，可以使用ConcurrentHashMap、CopyOnWriteArrayList等线程安全的集合类。

ThreadLocal

字节跳动, 中国交易与广告 - 实习, 一面

ThreadLocal 提供线程本地存储机制，为每个线程创建变量的独立副本，从而避免多线程环境下的数据竞争问题。

对 ThreadLocal 的操作有4种方法：

1. set(T value)：设置当前线程的局部变量值。
2. get()：获取当前线程的局部变量值。
3. remove()：移除当前线程的局部变量值。
4. initialValue()：提供初始值的方式，子类可以重写该方法。

应用场景

• 在 Web 应用中，存储用户的会话信息（即 Session）。
• 在数据库操作中，管理连接对象，确保每个线程使用独立的链接，避免冲突。

每个线程内部维护一个 ThreadLocalMap，用于存储私有变量。而 ThreadLocal 本身只是一个工具，用于操作 Thread 对象内部的 ThreadLocalMap。

当我们调用 threadLocal.set(value) 时，实际发生了以下事情

1. 获取当前线程 Thread 对象。
2. 获取当前 Thread 对象内部的 ThreadLocalMap，若没有则创建一个。
3. 以 threadLocal 对象本身作为 Key，要存储的 value 作为 Value，存放到当前线程的 ThreadLocalMap 中。

这里作为 key 的 ThreadLocal 对象，ThreadLocalMap 对它的引用是弱引用；而我们要存的 value（实际数据），ThreadLocalMap 对它的引用是强引用. 这也是后面会发生内存泄漏的核心原因.

当我们调用 threadLocal.get() 方法时，实际发生了以下事情

1. 获取当前线程的 Thread 对象。
2. 获取当前线程 Thread 对象内部的 ThreadLocalMap。
3. 以 threadLocal 对象本身作为 Key，从 ThreadLocalMap 中取出对应的 Value 并返回。

当 ThreadLocal 对象被回收后，键值对中的 key 就变成了 null，但 value 的值作为强引用，仍然被键值对引用，仍然存在，且无法被清理，最终造成内存泄漏。

因此在登出等操作后，一定要手动 remove() 掉 ThreadLocalMap 中的键值对。

设置成弱引用的原因也比较简单。假设 ThreadLocalMap 中对 ThreadLocal 的引用的强引用。若 ThreadLocal 在外部已经没有被任何地方引用了，即应该被回收了，但是由于 ThreadLocalMap 中的键值对仍然强引用着这个 ThreadLocal 对象，导致该对象永远无法被回收，就会导致这个对象本身的内存泄漏。

其他

• ThreadLocalMap 使用数组结构，通过线性探测法解决哈希冲突。
• 扩容前会先清理无效条目，当填充率达 2/3 时进行扩容（容量翻倍并重哈希）。
• 默认情况下，子线程无法继承父线程的 ThreadLocal，需使用 InheritableThreadLocal 实现继承，其原理是在创建子线程时拷贝父线程的 InheritableThreadLocalMap。

线程池

基本信息

线程池是用来管理和复用线程的工具，可以提高系统的性能和资源利用率。Java 中线程池的核心实现为 ThreadPoolExecutor，并提供了 Executor 框架来简化线程池的创建和管理（但不推荐使用）。

在创建线程时可以显式指定线程组

public Thread(ThreadGroup group, Runnable target);
public Thread(ThreadGroup group, String name);
public Thread(ThreadGroup group, Runnable target, String name);

工作流程

1. 线程池通过 submit() 或 execute() 方法接收任务。
2. 线程池首先检查核心线程数是否已满，若未满则创建新线程执行任务。
3. 若核心线程数已满，则将任务放入任务队列中。
4. 若任务队列已满且线程数未达最大值，则创建新线程执行任务。
5. 若任务队列已满且线程数已达最大值，则根据拒绝策略处理任务。
6. 线程执行完任务后，若线程数超过核心线程数且空闲时间超过 keepAliveTime，则终止该线程

graph TD
    A[execute或submit] --> B{核心线程数未满?}
    
    B -->|是| C[创建新线程执行任务]
    B -->|否| D{任务队列未满?}
    
    D -->|是| E[将任务放入任务队列]
    D -->|否| F{最大线程数未满?}
    
    F -->|是| G[创建新线程执行任务]
    F -->|否| H[执行拒绝策略]

核心参数

• corePoolSize：核心线程数，线程池中始终保持的线程数量。
• maximumPoolSize：最大线程数，线程池中允许的最大线程数量。
• workQueue：任务队列，用于存储等待执行的任务。
  以上三个参数最重要，基本决定了线程池对于任务的处理策略。
• keepAliveTime：线程空闲时间，超过该时间的非核心线程将被终止。
• unit：时间单位，keepAliveTime 的时间单位。
• threadFactory：线程工厂，用于创建新线程。
• handler：拒绝策略，当任务无法执行时的处理方式。

提交任务

• execute() 用于提交不需要返回值的任务。当任务执行过程中抛出异常时，会导致执行该任务的线程中止，但线程池会创建一个新的工作线程来替代它，确保线程池的正常运行。
• submit() 用于提交需要返回值的任务，返回一个 Future 对象。当任务执行过程中发生异常时，异常会被封装在 Future 对象中，不会立即抛出。只有在调用 Future.get() 方法时，才会抛出执行时遇到的异常。

因此，使用 submit() 方法时，务必通过 Future.get() 来获取执行结果或捕获可能发生的异常，否则任务中的异常可能会被忽略。

Future 类用于异步计算。其主要实现了以下4种功能

• boolean cancel() 取消任务
• boolean isCancelled() 查看是否已取消任务
• boolean isDone() 查看是否已经执行完成
• V get(long timeout, TimeUnit unit) 获取任务执行结果，超时未完成则抛出异常

然而，Future 类在实际使用过程中也存在一些局限，比如不支持异步任务的编排组合，获取结果的 get() 方法为阻塞调用等。因此在 Java 8 中引入了 CompletableFuture 类，提供了函数式编程和异步编排组合等功能。

若一个任务需要等待多个任务执行之后执行，这种需求就很适合通过 CompletableFuture 实现。

// T1
CompletableFuture<Void> futureT1 = CompletableFuture.runAsync(() -> {
    System.out.println("T1 is executing. Current time：" + DateUtil.now());
    // 模拟耗时操作
    ThreadUtil.sleep(1000);
});
// T2
CompletableFuture<Void> futureT2 = CompletableFuture.runAsync(() -> {
    System.out.println("T2 is executing. Current time：" + DateUtil.now());
    ThreadUtil.sleep(1000);
});

// 使用allOf()方法合并T1和T2的CompletableFuture，等待它们都完成
CompletableFuture<Void> bothCompleted = CompletableFuture.allOf(futureT1, futureT2);
// 当T1和T2都完成后，执行T3
bothCompleted.thenRunAsync(() -> System.out.println("T3 is executing after T1 and T2 have completed.Current time：" + DateUtil.now()));
// 等待所有任务完成，验证效果
ThreadUtil.sleep(3000);

使用该类时，可以使用 whenComplete 方法在任务完成时触发回调函数，以正确处理异常；使用 exceptionally 处理异常并重新抛出；使用 handle 方法处理正常的结果和异常，并返回一个新结果。

CompletableFuture 默认使用 ForkJoinPool 作为线程池。

拒绝策略

• AbortPolicy：默认策略，抛出 RejectedExecutionException 异常

• CallerRunsPolicy：调用者运行策略，任务由调用者线程执行

• DiscardPolicy：丢弃策略，直接丢弃任务，不抛出异常

• DiscardOldestPolicy：丢弃最旧任务策略，丢弃任务队列中最旧的任务，然后尝试执行当前任务

CallerRunsPolicy 适用于保证任何一个任务请求都要被执行的时候。但是，如果走到 CallerRunsPolicy 的任务是个非常耗时的任务，且处理提交任务的线程是主线程，可能会导致主线程阻塞，影响程序的正常进行，甚至导致线程池阻塞，最后堆到其他线程的任务越来越多，导致 OOM。

因此一般是把线程池外的任务存到 MQ 或阻塞队列中。此时需要实现 RejectedExecutionHandler 接口自定义拒绝策略。

关闭方式

• shutdown()：等待所有任务执行完毕后关闭
• shutdownNow()：立即关闭，尝试中断所有正在执行的任务，并忽略队列中未执行的任务

线程数安排多少个比较合适呢？一般需要分析线程池执行的任务类型是CPU密集型还是IO密集型。

• CPU 密集型：线程数 ≈ CPU核心数 + 1
  目标是最小化线程上下文切换，+1 作为备用线程处理可能的阻塞

• IO 密集型：线程数 ≈ CPU核心数 × 2～3
  因线程经常在IO操作时阻塞，可配置更多线程提高CPU利用率

实际配置需结合业务场景：CPU使用率低可能需增加线程数，CPU使用率高但吞吐量低可能需减少线程数。

常见的线程池

常见的线程池有4种

• FixedThreadPool：固定大小线程池，适用于任务量较大且任务执行时间较长的场景。如IO密集型任务、数据库连接池等。
  其线程池大小是固定的，默认使用 LinkedBlockingQueue 作为任务队列。其缺点是任务队列默认无界，可能会导致内存溢出（或者说会导致 OOM）。
• CachedThreadPool：可缓存线程池，适用于任务量较大且任务执行时间较短的场景。如短时间内大量的文件处理或网络请求等。
  其线程池大小不固定。空闲线程超过60秒会被终止，默认使用 SynchronousQueue 作为任务队列。其优点是线程池可以根据任务量动态调整大小，缺点是线程数没有上限，可能会创建大量线程，导致系统资源耗尽。
• SingleThreadExecutor：单线程线程池，适用于需要顺序执行任务的场景。如日志处理、定时任务等。
  线程池只有一个线程，保证任务按顺序执行，默认使用 LinkedBlockingQueue 作为任务队列。其缺点是单线程可能成为性能瓶颈。
• ScheduledThreadPool：定时任务线程池，适用于需要定时或周期性执行任务的场景。如定时数据备份、定时清理缓存等。
  线程池大小可配置，支持定时或周期性任务执行。默认使用 DelayedWorkQueue 作为任务队列。其优点是支持定时和周期性任务，缺点是线程数没有上限，可能会创建大量线程，导致系统资源耗尽。

这些线程池的默认拒绝策略均为 AbortPolicy。

状态管理

线程池的异常处理常见有四种方式

1. 最简单的，try-catch处理。
2. 使用Future对象的get()方法获取异常。
   建议使用submit()的项目使用这种方式。
3. 自定义线程池，重写afterExecute()方法处理异常。
   建议想要全局处理异常的项目使用这种方式。
4. 使用UncaughtExceptionHandler处理未捕获异常。
   建议使用execute()的项目使用这种方式。

线程池有5种状态，并且它们之间严格按照状态流转规则流转。

graph TD
    A[RUNNING] --> |shutdown|B[SHUTDOWN]
    A-->|shutdownNow|C[STOP]
    B --> |队列为空且工作线程数为0|D[TIDYING]
    C --> |工作线程数为0|D
    D --> |terminated|E[TERMINATED]

RUNNING 状态的线程池可以接收新任务，并处理阻塞队列中的任务；SHUTDOWN 状态的线程池不会接收新任务，但会处理阻塞队列中的任务；STOP 状态的线程池不会接收新任务，也不会处理阻塞队列中的任务，并且会尝试中断正在执行的任务；TIDYING 状态表示所有任务已经终止；TERMINATED 状态表示线程池完全关闭，所有线程销毁。

使用线程池提供的setter()方法就可以修改线程池的参数。

需要注意的是，调用 setCorePoolSize() 时如果新的核心线程数比原来的大，线程池会创建新的线程；如果更小，线程池不会立即销毁多余的线程，除非有空闲线程超过 keepAliveTime。
当然了，还可以利用 Nacos 配置中心，或者实现自定义的线程池，监听参数变化去动态调整参数。

应用

线程资源必须通过线程池提供，不应该在应用中自行显式创建线程。并且，线程池不应该使用 Executor 创建，更应该通过 ThreadPoolExecutor 构造函数的方式构建。因为使用 Executor 创建的四种线程池都有或多或少的问题。

ThreadPoolExecutor 默认不会回收核心线程。要是实在需要回收核心线程，可以考虑将 allowCoreThreadTimeout(boolean value) 设置为 true，这样就会回收空闲的核心线程了，其时间间隔由 keepAliveTime 指定。

核心线程在空闲时，会处于 WAITING 状态，当队列中有可用任务时，会唤醒被阻塞的线程，线程的状态会由 WAITING 状态变为 RUNNABLE 状态，之后去执行对应的任务。

要设计一个根据任务的优先级来执行的线程池，可以考虑使用 PriorityBlockingQueue 作为任务队列。当然，使用优先队列实现对任务的排序，传入其中的任务必须是具备排序能力的，因此需要提交到线程池的任务实现 Comparable 接口或 Comparator 接口。并且为了避免 OOM 问题，最好重写一下 offer 方法的逻辑，在元素数量超过指定值时返回 false。