本文旨在探讨如何在Java中高效且正确地创建一个包含多个`BlockingQueue
在Java并发编程中，我们经常需要处理线程安全的数据结构。当需要管理一组独立的、线程安全的队列时，例如一个由多个BlockingQueue组成的列表，并确保对该列表本身的访问也是同步的，这就需要特别注意。本文将详细介绍如何构建一个List<BlockingQueue<AtomicInteger>>，使其不仅内部的BlockingQueue是线程安全的，而且对该列表的添加、删除等操作也具备同步性。
理解ArrayList容量与列表内容的区别
在尝试构建一个包含预定数量BlockingQueue的列表时，一个常见的误区是使用ArrayList的带容量参数的构造函数，例如 new ArrayList<ArrayBlockingQueue<AtomicInteger>>(15)。这个构造函数仅仅是预分配了底层数组的容量，以减少后续元素添加时的扩容开销，但它并不会向列表中添加任何实际的元素。因此，调用 listOfQueues.size() 仍然会返回 0。
要解决这个问题，我们必须手动地向列表中添加BlockingQueue实例。此外，需要注意的是，像ArrayBlockingQueue这样的BlockingQueue实现，其构造函数通常也需要指定队列自身的容量。
立即学习“Java免费学习笔记（深入）”；
方案一：使用传统for循环构建同步列表
传统的for循环方法直观且易于理解。它分两步完成：首先，创建一个普通的ArrayList并使用循环向其中添加指定数量的BlockingQueue实例；然后，使用Collections.synchronizedList()方法将这个列表包装成一个同步列表。
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class SynchronizedBlockingQueueListForLoop {
public static void main(String[] args) {
// 定义每个BlockingQueue的容量
final int queueCapacity = 10;
// 定义列表中BlockingQueue的数量
final int numberOfQueues = 15;
// 步骤1: 创建一个普通的ArrayList来存放BlockingQueue实例
List<BlockingQueue<AtomicInteger>> tempListOfQueues = new ArrayList<>();
// 步骤2: 使用for循环向列表中添加指定数量的ArrayBlockingQueue实例
for (int i = 0; i < numberOfQueues; i++) {
// 注意：ArrayBlockingQueue的构造函数需要指定容量
tempListOfQueues.add(new ArrayBlockingQueue<>(queueCapacity));
}
// 步骤3: 使用Collections.synchronizedList()将列表包装成一个同步列表
List<BlockingQueue<AtomicInteger>> synchronizedListOfQueues =
Collections.synchronizedList(tempListOfQueues);
System.out.println("使用for循环创建的同步列表大小: " + synchronizedListOfQueues.size());
// 验证列表中的元素类型
if (!synchronizedListOfQueues.isEmpty()) {
System.out.println("第一个队列的类型: " + synchronizedListOfQueues.get(0).getClass().getSimpleName());
}
}
}注意事项：
- Collections.synchronizedList()返回的是一个包装器，它对原始列表的所有操作都进行了同步。这意味着对synchronizedListOfQueues的任何修改（如add, remove）或结构性访问（如iterator）都将是线程安全的。
- 然而，这仅仅是列表本身的同步，列表内部的BlockingQueue实例仍然是独立的线程安全对象。对BlockingQueue内部元素的操作（如put, take）由BlockingQueue自身保证线程安全。
方案二：使用Java Stream API构建同步列表
Java 8引入的Stream API提供了一种更函数式、更简洁的方式来创建和处理集合。我们可以利用Stream.generate()来生成BlockingQueue实例，然后使用limit()限制数量，最后通过collectingAndThen()收集并同步列表。
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class SynchronizedBlockingQueueListStream {
public static void main(String[] args) {
// 定义每个BlockingQueue的容量
final int queueCapacity = 10;
// 定义列表中BlockingQueue的数量
final int numberOfQueues = 15;
// 使用Stream API生成、收集并同步列表
List<BlockingQueue<AtomicInteger>> synchronizedListOfQueues =
Stream.generate(() -> new ArrayBlockingQueue<AtomicInteger>(queueCapacity)) // 生成ArrayBlockingQueue实例的流
.limit(numberOfQueues) // 限制生成的数量
.collect(Collectors.collectingAndThen( // 收集并进行后处理
Collectors.toList(), // 先收集成一个普通的List
Collections::synchronizedList // 然后将List包装成同步列表
));
System.out.println("使用Stream API创建的同步列表大小: " + synchronizedListOfQueues.size());
// 验证列表中的元素类型
if (!synchronizedListOfQueues.isEmpty()) {
System.out.println("第一个队列的类型: " + synchronizedListOfQueues.get(0).getClass().getSimpleName());
}
}
}Stream API方法的优势：
- 简洁性： 代码更加紧凑，意图表达更清晰。
- 链式操作： 允许将多个操作（生成、限制、收集、同步）以链式方式组合，提高了可读性。
- 不可变性倾向： collectingAndThen在创建完中间列表后进行同步包装，更符合函数式编程中不可变数据的理念。
总结
无论是采用传统的for循环还是现代的Stream API，核心思想都是一致的：
- 正确初始化列表内容： ArrayList的构造函数只设置容量，不添加元素。必须通过循环或流式操作显式地添加BlockingQueue实例。
- 为BlockingQueue指定容量： 像ArrayBlockingQueue这样的有界阻塞队列，在构造时需要指定其内部容量。
- 列表同步： 使用Collections.synchronizedList()来确保对整个列表（而非其内部元素）的并发访问是线程安全的。
选择哪种方法取决于个人偏好和项目上下文。对于熟悉Stream API的开发者，第二种方法可能更具吸引力；而对于需要更明确控制流程的场景，for循环方法依然是稳健的选择。理解这些基本概念和实现细节，对于在Java中构建健壮的并发数据结构至关重要。
