本文将深入探讨在Java中如何正确构建一个包含多个`BlockingQueue
在Java并发编程中，我们经常需要处理线程安全的数据结构。一个常见的需求是创建一个包含多个BlockingQueue实例的列表，并确保这个列表本身也是线程安全的。然而，在尝试初始化这样的数据结构时，开发者可能会遇到一些常见的误解。
理解ArrayList的容量与大小
许多开发者在初始化ArrayList时，可能会使用带容量参数的构造函数，例如 new ArrayList<>(15)，并期望这会创建一个包含15个空元素的列表。但实际上，这个构造函数仅仅是为ArrayList的底层数组预分配了内存空间，以优化后续的元素添加操作，而不会改变列表的实际大小（size）。这意味着，new ArrayList<>(15)创建的列表其size()仍然是0，因为它不包含任何元素。
要构建一个包含特定数量BlockingQueue实例的列表，我们需要显式地向列表中添加这些队列对象。此外，BlockingQueue是一个接口，需要选择其具体实现，例如ArrayBlockingQueue。ArrayBlockingQueue在其构造函数中也需要指定一个容量，这代表了该队列能够存储的最大元素数量。
立即学习“Java免费学习笔记（深入）”；
接下来，我们将介绍两种有效的方法来构建一个同步的、预填充了BlockingQueue实例的列表。
方法一：使用Java 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.Collections;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class SynchronizedBlockingQueueListGenerator {
public static void main(String[] args) {
final int queueCapacity = 10; // 每个BlockingQueue的容量
final int numberOfQueues = 5; // 列表中BlockingQueue的数量
// 使用Stream API生成并同步列表
List<BlockingQueue<AtomicInteger>> listOfQueues =
Stream.generate(() -> new ArrayBlockingQueue<AtomicInteger>(queueCapacity))
.limit(numberOfQueues)
.collect(Collectors.collectingAndThen(
Collectors.toList(),
Collections::synchronizedList
));
System.out.println("Stream API生成的列表大小: " + listOfQueues.size());
// 验证列表中的队列是否可操作
listOfQueues.forEach(queue -> {
System.out.println("队列容量: " + ((ArrayBlockingQueue<AtomicInteger>)queue).remainingCapacity());
try {
queue.put(new AtomicInteger(1));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
}代码解析：
- Stream.generate(() -> new ArrayBlockingQueue<AtomicInteger>(queueCapacity)): 创建一个无限流，每次调用都生成一个新的ArrayBlockingQueue实例。每个ArrayBlockingQueue都被初始化为指定的queueCapacity。
- .limit(numberOfQueues): 将无限流截断，使其只生成指定数量的BlockingQueue实例。
- .collect(Collectors.collectingAndThen(Collectors.toList(), Collections::synchronizedList)): 这是关键一步。
- Collectors.toList()：首先将流中的所有BlockingQueue实例收集到一个标准的ArrayList中。
- Collections::synchronizedList：然后，collectingAndThen会接收这个ArrayList，并对其应用Collections.synchronizedList方法，返回一个线程安全的列表视图。
方法二：使用传统的for循环
对于不熟悉Stream API的开发者，或者在简单场景下，使用传统的for循环迭代添加元素也是一个清晰直观的选择。
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.Collections;
public class SynchronizedBlockingQueueListGeneratorTraditional {
public static void main(String[] args) {
final int queueCapacity = 10; // 每个BlockingQueue的容量
final int numberOfQueues = 5; // 列表中BlockingQueue的数量
// 使用for循环生成列表
List<ArrayBlockingQueue<AtomicInteger>> tempListOfQueues = new ArrayList<>();
for (int i = 0; i < numberOfQueues; i++) {
tempListOfQueues.add(new ArrayBlockingQueue<>(queueCapacity));
}
// 将生成的列表包装成同步列表
List<BlockingQueue<AtomicInteger>> synchronizedListOfQueues =
Collections.synchronizedList(tempListOfQueues);
System.out.println("for循环生成的列表大小: " + synchronizedListOfQueues.size());
// 验证列表中的队列是否可操作
synchronizedListOfQueues.forEach(queue -> {
System.out.println("队列容量: " + ((ArrayBlockingQueue<AtomicInteger>)queue).remainingCapacity());
try {
queue.put(new AtomicInteger(2));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
}代码解析：
- List<ArrayBlockingQueue<AtomicInteger>> tempListOfQueues = new ArrayList<>();: 首先创建一个普通的ArrayList来存储ArrayBlockingQueue实例。
- for (int i = 0; i < numberOfQueues; i++) { tempListOfQueues.add(new ArrayBlockingQueue<>(queueCapacity)); }: 通过循环，按照指定的数量创建ArrayBlockingQueue实例，并逐一添加到tempListOfQueues中。
- List<BlockingQueue<AtomicInteger>> synchronizedListOfQueues = Collections.synchronizedList(tempListOfQueues);: 在所有队列都添加到tempListOfQueues之后，使用Collections.synchronizedList方法将其包装成一个线程安全的列表视图。
注意事项与最佳实践
- 同步范围： Collections.synchronizedList()方法只确保了对列表结构（如添加、删除元素，或者获取迭代器）的并发访问是线程安全的。它不会自动同步列表中每个BlockingQueue实例内部的操作（如put()、take()）。BlockingQueue的实现（如ArrayBlockingQueue）本身就是线程安全的，因此对其元素的并发操作无需额外同步。
- BlockingQueue容量： ArrayBlockingQueue是一个有界队列，其构造函数必须指定容量。请根据实际需求合理设置这个容量，因为它决定了队列能够存储的最大元素数量。
- 泛型类型： 在示例中，我们使用了BlockingQueue<AtomicInteger>。AtomicInteger是Java并发包提供的一个原子类，用于在多线程环境下安全地操作整数，避免了手动同步。如果队列中存储的是其他类型的对象，且这些对象本身不是线程安全的，那么在访问这些对象时可能需要额外的同步措施。
- 选择合适的BlockingQueue实现： 除了ArrayBlockingQueue，Java还提供了LinkedBlockingQueue（链表实现，默认无界或可指定容量）、PriorityBlockingQueue（优先级队列）、DelayQueue（延迟队列）等。根据应用场景选择最适合的BlockingQueue实现。
总结
正确构建一个包含多个BlockingQueue实例的同步列表，关键在于理解ArrayList的容量与大小的区别，并显式地填充列表。无论是采用现代的Stream API结合collectingAndThen的函数式方法，还是传统的for循环迭代方法，都能有效地实现这一目标。最终生成的同步列表将为并发应用程序提供一个稳定且线程安全的基础结构，但请务必记住Collections.synchronizedList的同步范围限制，并结合BlockingQueue自身的线程安全特性进行编程。
