本文深入探讨了在Java中如何正确初始化一个包含指定数量、且内部元素为`AtomicInteger`的`BlockingQueue`列表，并确保该列表本身是线程安全的。文章阐明了`ArrayList`容量与实际元素数量的区别，以及`BlockingQueue`实例化时的容量要求，提供了基于Stream API和传统for循环的两种高效且线程安全的实现方案，旨在帮助开发者避免常见误区，构建健壮的并发数据结构。
在Java并发编程中，我们经常需要处理线程安全的数据结构。当需要一个包含多个并发队列的列表时，例如List
理解ArrayList的容量与大小
一个常见的误解是，当使用new ArrayList
BlockingQueue的实例化与容量
BlockingQueue是一个接口，其具体实现如ArrayBlockingQueue在构造时通常需要指定其自身的容量。这个容量定义了队列可以存储的最大元素数量。例如，new ArrayBlockingQueue(capacity)会创建一个固定容量的阻塞队列。因此，在创建BlockingQueue实例并将其添加到列表中时，需要为每个队列指定合适的容量。
立即学习“Java免费学习笔记（深入）”；
构建同步的BlockingQueue列表
为了得到一个线程安全的List，我们通常使用Collections.synchronizedList()方法对一个普通的List进行包装。这个方法会返回一个线程安全的列表视图，其所有操作都会通过内部锁进行同步。
以下是两种常用的方法来构建一个包含指定数量的BlockingQueue，并使其列表本身是同步的：
方法一：使用Stream API与collectingAndThen()
Java 8引入的Stream API提供了一种声明式且简洁的方式来生成和收集数据。结合Collectors.collectingAndThen()，我们可以在收集元素到列表之后，再对这个列表进行额外的转换（例如使其同步）。
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;
import java.util.Collections;
public class SynchronizedBlockingQueueList {
public static void main(String[] args) {
final int queueCapacity = 10; // 每个BlockingQueue的容量
final int numberOfQueues = 5; // 列表中BlockingQueue的数量
// 使用Stream API生成并同步列表
List> listOfQueues =
Stream.generate(() -> new ArrayBlockingQueue(queueCapacity))
.limit(numberOfQueues)
.collect(Collectors.collectingAndThen(
Collectors.toList(),
Collections::synchronizedList
));
System.out.println("使用Stream API创建的列表大小: " + listOfQueues.size());
// 验证每个队列的容量
listOfQueues.forEach(queue -> {
System.out.println(" 队列类型: " + queue.getClass().getSimpleName() + ", 容量: " + ((ArrayBlockingQueue)queue).remainingCapacity() + queue.size());
});
// 示例：向第一个队列添加元素
if (!listOfQueues.isEmpty()) {
listOfQueues.get(0).offer(new AtomicInteger(1));
System.out.println("第一个队列的元素数量: " + listOfQueues.get(0).size());
}
}
} 代码解析：
- Stream.generate(() -> new ArrayBlockingQueue
(queueCapacity)): 创建一个无限流，每个元素都是一个新的ArrayBlockingQueue实例，其容量由queueCapacity指定。 - .limit(numberOfQueues): 将无限流截断，只取前numberOfQueues个元素。
- Collectors.collectingAndThen(Collectors.toList(), Collections::synchronizedList): 这是一个复合收集器。
- Collectors.toList(): 作为下游收集器，将流中的元素收集到一个普通的ArrayList中。
- Collections::synchronizedList: 作为转换函数，在ArrayList创建完成后，将其包装成一个线程安全的List。
方法二：使用传统的for循环
对于不熟悉Stream API或偏好更直观的循环结构的开发者，传统的for循环也是一个清晰且有效的选择。
import java.util.ArrayList;
import java.util.List;
import java.util.Collections;
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) {
final int queueCapacity = 10; // 每个BlockingQueue的容量
final int numberOfQueues = 5; // 列表中BlockingQueue的数量
// 先创建一个普通的ArrayList
List> tempListOfQueues = new ArrayList<>();
// 使用for循环手动添加BlockingQueue实例
for (int i = 0; i < numberOfQueues; i++) {
tempListOfQueues.add(new ArrayBlockingQueue<>(queueCapacity));
}
// 将普通的ArrayList包装成线程安全的List
List> synchronizedListOfQueues =
Collections.synchronizedList(tempListOfQueues);
System.out.println("使用for循环创建的列表大小: " + synchronizedListOfQueues.size());
// 验证每个队列的容量
synchronizedListOfQueues.forEach(queue -> {
System.out.println(" 队列类型: " + queue.getClass().getSimpleName() + ", 容量: " + ((ArrayBlockingQueue)queue).remainingCapacity() + queue.size());
});
// 示例：向第二个队列添加元素
if (!synchronizedListOfQueues.isEmpty()) {
synchronizedListOfQueues.get(1).offer(new AtomicInteger(2));
System.out.println("第二个队列的元素数量: " + synchronizedListOfQueues.get(1).size());
}
}
} 代码解析：
- List
> tempListOfQueues = new ArrayList();: 首先创建一个普通的ArrayList，用于临时存储BlockingQueue实例。 - for (int i = 0; i (queueCapacity)); }: 循环numberOfQueues次，每次创建一个新的ArrayBlockingQueue实例并将其添加到tempListOfQueues中。
- List
> synchronizedListOfQueues = Collections.synchronizedList(tempListOfQueues);: 最后，将填充好的tempListOfQueues传递给Collections.synchronizedList()方法，得到一个线程安全的列表。
注意事项与最佳实践
- 线程安全性范围： Collections.synchronizedList()只保证对列表本身的结构性修改（如添加、删除元素）是线程安全的，以及对列表元素的访问是同步的。然而，列表中的每个BlockingQueue实例的内部操作（如put(), take(), `offer(), poll()) 本身就是线程安全的，这是BlockingQueue接口的设计特性。因此，无需对BlockingQueue的内部操作再进行额外的同步。
- 选择BlockingQueue实现： ArrayBlockingQueue是固定容量的阻塞队列。根据实际需求，也可以选择其他BlockingQueue的实现，例如LinkedBlockingQueue（可选容量，默认无界）或PriorityBlockingQueue（优先级队列）。
- 容量规划： 仔细规划BlockingQueue的容量（queueCapacity）和列表中BlockingQueue的数量（numberOfQueues）。不合理的容量可能导致性能瓶颈或资源浪费。
- var关键字： 在Java 10及更高版本中，可以使用var关键字来简化局部变量的类型声明，如示例代码所示，提高代码简洁性。
总结
正确地初始化一个同步的BlockingQueue列表，关键在于理解ArrayList容量与大小的区别，以及BlockingQueue实例化时的容量要求。通过Stream API的collectingAndThen()方法或传统的for循环，我们都能有效地构建出符合需求的线程安全数据结构。选择哪种方法取决于个人偏好和项目代码风格。无论哪种方式，最终目标都是为了在并发环境中安全、高效地管理和操作这些队列。
