Java 并发队列（Concurrent Queue）全解析

简介

在 Java 多线程编程中，线程安全是一个至关重要的问题。当多个线程需要共享数据时，如果没有合适的同步机制，就可能会出现数据不一致、竞态条件等问题。Java 中的并发队列（Concurrent Queue）就是为了解决这些问题而设计的，它提供了一种线程安全的方式来在多个线程之间传递数据。本文将详细介绍 Java 并发队列的基础概念、使用方法、常见实践以及最佳实践，帮助读者深入理解并高效使用 Java 并发队列。

基础概念

什么是并发队列

并发队列是一种线程安全的队列，它允许在多线程环境下进行高效的入队（enqueue）和出队（dequeue）操作。在 Java 中，并发队列通常位于 java.util.concurrent 包下，主要有两种类型：阻塞队列（Blocking Queue）和非阻塞队列（Non-blocking Queue）。

阻塞队列和非阻塞队列

阻塞队列（Blocking Queue）：当队列满时，尝试入队的线程会被阻塞，直到队列有空间；当队列空时，尝试出队的线程会被阻塞，直到队列有元素。常见的阻塞队列有 ArrayBlockingQueue、LinkedBlockingQueue 等。
非阻塞队列（Non-blocking Queue）：入队和出队操作不会阻塞线程，如果操作无法立即完成，会返回一个特定的值（如 null）。常见的非阻塞队列有 ConcurrentLinkedQueue。

使用方法

阻塞队列示例：`ArrayBlockingQueue`

import java.util.concurrent.ArrayBlockingQueue;

public class BlockingQueueExample {
    public static void main(String[] args) {
        // 创建一个容量为 5 的 ArrayBlockingQueue
        ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);

        // 生产者线程
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    queue.put(i);
                    System.out.println("Produced: " + i);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 消费者线程
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    Integer item = queue.take();
                    System.out.println("Consumed: " + item);
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 启动线程
        producer.start();
        consumer.start();

        // 等待线程结束
        try {
            producer.join();
            consumer.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

非阻塞队列示例：`ConcurrentLinkedQueue`

import java.util.concurrent.ConcurrentLinkedQueue;

public class NonBlockingQueueExample {
    public static void main(String[] args) {
        // 创建一个 ConcurrentLinkedQueue
        ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();

        // 生产者线程
        Thread producer = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                queue.offer(i);
                System.out.println("Produced: " + i);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        // 消费者线程
        Thread consumer = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                Integer item = queue.poll();
                if (item != null) {
                    System.out.println("Consumed: " + item);
                }
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        // 启动线程
        producer.start();
        consumer.start();

        // 等待线程结束
        try {
            producer.join();
            consumer.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

常见实践

生产者 - 消费者模式

生产者 - 消费者模式是并发队列最常见的应用场景之一。在该模式中，生产者线程负责生产数据并将其放入队列，消费者线程负责从队列中取出数据并进行处理。上述示例代码就是生产者 - 消费者模式的实现。

任务调度

并发队列可以用于任务调度，例如将待执行的任务放入队列，然后由多个工作线程从队列中取出任务并执行。

import java.util.concurrent.ArrayBlockingQueue;

public class TaskScheduler {
    private final ArrayBlockingQueue<Runnable> taskQueue;

    public TaskScheduler(int capacity) {
        taskQueue = new ArrayBlockingQueue<>(capacity);
    }

    public void submitTask(Runnable task) {
        try {
            taskQueue.put(task);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public void startWorkers(int numWorkers) {
        for (int i = 0; i < numWorkers; i++) {
            Thread worker = new Thread(() -> {
                try {
                    while (true) {
                        Runnable task = taskQueue.take();
                        task.run();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            worker.start();
        }
    }

    public static void main(String[] args) {
        TaskScheduler scheduler = new TaskScheduler(10);
        scheduler.startWorkers(3);

        // 提交任务
        for (int i = 0; i < 5; i++) {
            final int taskId = i;
            scheduler.submitTask(() -> {
                System.out.println("Executing task: " + taskId);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
    }
}

最佳实践

选择合适的队列类型

根据具体的应用场景选择合适的队列类型。如果需要在队列满或空时阻塞线程，使用阻塞队列；如果不需要阻塞线程，使用非阻塞队列。

避免队列容量过大或过小

队列容量过大会占用过多的内存，过小则可能导致频繁的阻塞和唤醒操作，影响性能。根据实际情况合理设置队列容量。

异常处理

在使用并发队列时，要注意处理可能出现的异常，如 InterruptedException。当线程被中断时，要正确处理中断状态。

小结

Java 并发队列是多线程编程中非常重要的工具，它提供了一种线程安全的方式来在多个线程之间传递数据。本文介绍了并发队列的基础概念、使用方法、常见实践以及最佳实践。通过合理使用并发队列，可以提高多线程程序的性能和可靠性。

参考资料

《Effective Java》