Java 延迟队列与线程池
2021-09-177.5k 阅读
Java 延迟队列（DelayQueue）概述
在Java的并发包 java.util.concurrent 中，DelayQueue 是一个无界的阻塞队列，它的元素只有在其延迟期满时才能从队列中取走。这使得 DelayQueue 非常适合实现一些具有延迟执行特性的任务，例如定时任务、缓存过期清理等场景。
DelayQueue 中的元素必须实现 Delayed 接口，该接口定义了两个方法：
public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
int compareTo(Delayed o);
}
getDelay(TimeUnit unit)方法返回距离该元素到期还剩余的时间，时间单位由unit指定。compareTo(Delayed o)方法用于比较两个Delayed对象，通常根据到期时间来进行比较，以便在队列中正确排序。
代码示例 - 简单的延迟任务
下面通过一个简单的代码示例来展示如何使用 DelayQueue。假设我们有一个简单的任务，任务有一个执行时间点，在该时间点到达后任务才会被执行。
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
class DelayedTask implements Delayed {
private final long delayTime;
private final long expireTime;
private final String taskName;
public DelayedTask(String taskName, long delayTime) {
this.taskName = taskName;
this.delayTime = delayTime;
this.expireTime = System.currentTimeMillis() + delayTime;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(expireTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed o) {
DelayedTask other = (DelayedTask) o;
return Long.compare(this.expireTime, other.expireTime);
}
public void execute() {
System.out.println("Task " + taskName + " is executed.");
}
}
public class DelayQueueExample {
public static void main(String[] args) {
DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();
delayQueue.add(new DelayedTask("Task1", 2000)); // 2秒后执行
delayQueue.add(new DelayedTask("Task2", 1000)); // 1秒后执行
Thread worker = new Thread(() -> {
while (true) {
try {
DelayedTask task = delayQueue.take();
task.execute();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
});
worker.start();
try {
Thread.sleep(3000);
worker.interrupt();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
在上述代码中：
DelayedTask类实现了Delayed接口，它有一个任务名称taskName，延迟时间delayTime和到期时间expireTime。getDelay方法计算剩余延迟时间，compareTo方法根据到期时间比较任务。DelayQueueExample类中创建了一个DelayQueue，并添加了两个延迟任务。worker线程不断从队列中取出到期任务并执行。主线程睡眠3秒后中断worker线程，结束程序。
DelayQueue 的内部实现原理
DelayQueue 内部是基于 PriorityQueue 实现的，PriorityQueue 是一个基于堆的无界优先级队列。在 DelayQueue 中，元素按照它们的到期时间排序，最早到期的元素在队列头部。
当调用 add(E e) 方法（或者 offer(E e) 方法）向 DelayQueue 中添加元素时，实际上是调用了 PriorityQueue 的 offer(E e) 方法，PriorityQueue 会将新元素插入到堆中合适的位置，以维护堆的顺序性。
当调用 take() 方法从 DelayQueue 中取出元素时：
- 首先检查队列是否为空，如果为空则阻塞当前线程，直到有元素可用。
- 取出堆顶元素（即最早到期的元素），检查该元素是否到期。如果到期，则直接返回该元素；如果未到期，则计算剩余延迟时间，当前线程进入等待状态，等待剩余延迟时间过去后再次尝试获取元素。
Java 线程池（ThreadPool）概述
线程池是一种管理和复用线程的机制，它可以避免频繁创建和销毁线程带来的开销，提高应用程序的性能和资源利用率。在Java中，线程池的核心实现类是 ThreadPoolExecutor，位于 java.util.concurrent 包中。
ThreadPoolExecutor 有多个构造函数，其中最常用的一个构造函数如下：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
corePoolSize：核心线程数，线程池在正常情况下保持的线程数量，即使这些线程处于空闲状态，也不会被销毁（除非设置了allowCoreThreadTimeOut为true）。maximumPoolSize：最大线程数，线程池允许创建的最大线程数量。keepAliveTime：存活时间，当线程数大于核心线程数时，多余的空闲线程等待新任务的最长时间，超过这个时间就会被销毁。unit：存活时间的单位，如TimeUnit.SECONDS、TimeUnit.MILLISECONDS等。workQueue：任务队列，用于存放等待执行的任务。这是一个BlockingQueue，常见的实现有ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue等。threadFactory：线程工厂，用于创建新线程。通过线程工厂可以定制线程的一些属性，如线程名、线程优先级等。handler：拒绝策略，当任务队列已满且线程数达到最大线程数时，新任务到来时采取的策略。常见的拒绝策略有AbortPolicy（默认，直接抛出RejectedExecutionException）、CallerRunsPolicy（由调用者线程执行任务）、DiscardPolicy（丢弃任务）、DiscardOldestPolicy（丢弃队列中最老的任务，然后尝试重新提交新任务）。
代码示例 - 简单的线程池使用
下面是一个简单的线程池使用示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExample {
public static void main(String[] args) {
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(10);
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2, // 核心线程数
4, // 最大线程数
10, // 存活时间
TimeUnit.SECONDS,
workQueue,
r -> {
Thread thread = new Thread(r);
thread.setName("CustomThread-" + thread.getName());
return thread;
},
new ThreadPoolExecutor.CallerRunsPolicy()
);
for (int i = 0; i < 15; i++) {
int taskNumber = i;
executor.submit(() -> {
System.out.println("Task " + taskNumber + " is running on " + Thread.currentThread().getName());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
executor.shutdown();
try {
if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
executor.shutdownNow();
if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
System.err.println("Pool did not terminate");
}
}
} catch (InterruptedException ie) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
}
}
在上述代码中：
- 创建了一个
LinkedBlockingQueue作为任务队列，容量为10。 - 使用
ThreadPoolExecutor创建了一个线程池，核心线程数为2，最大线程数为4，存活时间为10秒。 - 通过
submit方法向线程池中提交15个任务，每个任务模拟执行2秒。 - 最后调用
shutdown方法关闭线程池，并通过awaitTermination方法等待线程池中的任务执行完毕。如果等待超时，则调用shutdownNow方法尝试停止未完成的任务。
线程池的工作原理
- 任务提交：当调用
execute(Runnable task)或submit(Callable<T> task)方法向线程池提交任务时，线程池会按照以下流程处理任务：- 如果当前线程数小于核心线程数，线程池会创建一个新线程来执行任务。
- 如果当前线程数达到核心线程数，任务会被放入任务队列
workQueue中等待执行。 - 如果任务队列已满，且当前线程数小于最大线程数，线程池会创建新线程来执行任务。
- 如果任务队列已满，且当前线程数达到最大线程数，根据设置的拒绝策略来处理新任务。
- 线程复用：线程池中的线程执行完一个任务后，并不会立即销毁，而是会从任务队列中获取下一个任务继续执行，从而实现线程的复用，减少线程创建和销毁的开销。
- 线程销毁：当线程池中的线程数大于核心线程数，且有线程空闲时间超过了
keepAliveTime，这些多余的线程会被销毁，直到线程数减少到核心线程数（除非设置了allowCoreThreadTimeOut为true，此时核心线程也可能会被销毁）。
DelayQueue 与线程池的结合使用
在实际应用中，我们可能会遇到需要将延迟任务提交到线程池中执行的场景。可以通过将延迟任务包装成 Runnable 或 Callable 对象，然后提交到线程池来实现。
代码示例 - 结合 DelayQueue 和线程池
import java.util.concurrent.*;
class DelayedTaskWrapper implements Runnable, Delayed {
private final long delayTime;
private final long expireTime;
private final Runnable task;
public DelayedTaskWrapper(Runnable task, long delayTime) {
this.task = task;
this.delayTime = delayTime;
this.expireTime = System.currentTimeMillis() + delayTime;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(expireTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed o) {
DelayedTaskWrapper other = (DelayedTaskWrapper) o;
return Long.compare(this.expireTime, other.expireTime);
}
@Override
public void run() {
task.run();
}
}
public class DelayQueueAndThreadPoolExample {
public static void main(String[] args) {
DelayQueue<DelayedTaskWrapper> delayQueue = new DelayQueue<>();
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>();
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2,
4,
10,
TimeUnit.SECONDS,
workQueue
);
delayQueue.add(new DelayedTaskWrapper(() -> System.out.println("Delayed Task 1 is executed"), 2000));
delayQueue.add(new DelayedTaskWrapper(() -> System.out.println("Delayed Task 2 is executed"), 1000));
Thread worker = new Thread(() -> {
while (true) {
try {
DelayedTaskWrapper task = delayQueue.take();
executor.submit(task);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
});
worker.start();
try {
Thread.sleep(3000);
worker.interrupt();
executor.shutdown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
在上述代码中：
DelayedTaskWrapper类既实现了Runnable接口，又实现了Delayed接口。它包装了一个普通的Runnable任务，并添加了延迟执行的功能。- 创建了一个
DelayQueue用于存放延迟任务包装对象，同时创建了一个ThreadPoolExecutor线程池。 worker线程不断从DelayQueue中取出到期的任务，并提交到线程池中执行。主线程睡眠3秒后中断worker线程，并关闭线程池。
应用场景分析
- 定时任务：可以使用
DelayQueue结合线程池来实现定时任务调度。将定时任务封装成延迟任务放入DelayQueue，由线程池中的线程执行这些任务。这种方式比使用Timer和TimerTask更灵活，且在多线程环境下更可靠。 - 缓存过期清理：在缓存系统中，当缓存项设置了过期时间时，可以使用
DelayQueue来管理这些缓存项。当缓存项到期时，从DelayQueue中取出并进行清理操作，同时可以利用线程池来提高清理效率。 - 消息延迟处理：在消息队列系统中，如果某些消息需要延迟处理，可以将这些消息封装成延迟任务放入
DelayQueue，然后由线程池中的线程在合适的时间处理这些消息。
性能优化与注意事项
- 线程池参数调优：合理设置线程池的核心线程数、最大线程数、任务队列容量等参数对于性能至关重要。如果核心线程数设置过小，可能导致任务长时间等待；如果设置过大，可能会消耗过多的系统资源。任务队列容量也需要根据实际情况进行调整，避免队列溢出。
- 内存管理：
DelayQueue是无界队列，如果不断添加延迟任务，可能会导致内存占用不断增加。在使用时需要注意对任务的清理和管理，避免内存泄漏。 - 异常处理：在线程池执行任务过程中，可能会抛出各种异常。需要合理设置拒绝策略和异常处理机制，确保应用程序的稳定性。例如，可以在
RejectedExecutionHandler中记录日志或者进行重试操作。 - 线程安全：
DelayQueue和线程池本身都是线程安全的，但在使用过程中，需要注意对共享资源的访问。如果任务中涉及到对共享资源的读写操作，需要使用合适的同步机制来保证线程安全。
通过深入理解和合理使用Java的延迟队列和线程池，可以开发出高效、稳定的并发应用程序，满足各种复杂的业务需求。无论是定时任务调度、缓存管理还是消息处理，它们都提供了强大的功能和灵活的解决方案。在实际应用中，需要根据具体场景进行性能优化和参数调优，以达到最佳的运行效果。同时，注意线程安全、内存管理和异常处理等方面，确保应用程序的健壮性和可靠性。