Java中的任务队列：从ThreadPoolExecutor到Disruptor

Java中的任务队列：从内置机制到现代框架

任务队列（Task Queues）是构建可扩展、高容错Java应用程序的基础模式。它们能够将业务逻辑与线程管理分离，平衡负载并异步处理任务。本文将深入解析Java中任务队列的主要实现方式、架构设计、代码示例及使用场景。

1. 内置机制：ThreadPoolExecutor 和 BlockingQueue

Java标准库（java.util.concurrent）提供了一套强大的任务队列工具。核心是ThreadPoolExecutor类，它管理线程池，并使用队列存储等待执行的任务。

基础示例：

import java.util.concurrent.*;

public class TaskQueueExample {    public static void main(String[] args) {        // 创建线程池：corePoolSize=2, maxPoolSize=4, 队列容量100        ThreadPoolExecutor executor = new ThreadPoolExecutor(            2, // corePoolSize            4, // maximumPoolSize            60, // keepAliveTime            TimeUnit.SECONDS,            new LinkedBlockingQueue<>(100) // 任务队列        );

        // 提交任务        for (int i = 0; i < 10; i++) {            final int taskId = i;            executor.submit(() -> {                System.out.println("任务 " + taskId + " 正在由线程 " +                                    Thread.currentThread().getName() + " 执行");                try {                    Thread.sleep(1000);                } catch (InterruptedException e) {                    Thread.currentThread().interrupt();                }            });        }

        // 正确关闭        executor.shutdown();    }}

ThreadPoolExecutor的队列类型：

• LinkedBlockingQueue — 基于链表的无界队列。适用于任务数量未知的系统。
• ArrayBlockingQueue — 基于数组的有界队列。可控制等待任务的最大数量。
• SynchronousQueue — 无容量的队列：每个put()等待take()。用于直接将任务传递给线程。
• PriorityBlockingQueue — 优先级队列。任务按指定比较器排序。

2. 基于队列的生产者-消费者架构

经典的生产者-消费者模式可以通过BlockingQueue轻松实现。生产者（producers）将任务放入队列，消费者（consumers）取出任务进行处理。

实现示例：

import java.util.concurrent.*;

public class ProducerConsumerExample {    private static final BlockingQueue<String> queue = new LinkedBlockingQueue<>(5);

    static class Producer implements Runnable {        @Override        public void run() {            try {                for (int i = 0; i < 10; i++) {                    String task = "任务-" + i;                    queue.put(task); // 队列满时阻塞                    System.out.println("已生产: " + task);                    Thread.sleep(100);                }            } catch (InterruptedException e) {                Thread.currentThread().interrupt();            }        }    }

    static class Consumer implements Runnable {        @Override        public void run() {            try {                while (true) {                    String task = queue.take(); // 队列空时阻塞                    System.out.printl