17370845950

本文详细介绍了如何在Java中创建一个包含多个`BlockingQueue`实例的同步列表。重点在于正确初始化`BlockingQueue`的容量，并确保列表本身是线程安全的。文章提供了使用Stream API和传统for循环两种实现方式，并附带示例代码，帮助开发者理解并解决创建同步阻塞队列列表时可能遇到的问题。

在Java并发编程中，BlockingQueue 是一种常用的线程安全的数据结构，它可以在多线程环境下安全地进行数据的生产和消费。有时我们需要一个列表来管理多个 BlockingQueue 实例，并且这个列表本身也需要是线程安全的。本文将介绍如何正确地创建一个包含多个 BlockingQueue 实例的同步列表。

问题分析

在尝试创建 List> 时，直接使用 Collections.synchronizedList(new ArrayList>(15)) 可能会导致一些问题。主要问题在于：

1. ArrayList 构造函数中的参数 15 指定的是底层数组的初始容量，而不是列表的初始大小。这意味着列表初始时仍然是空的，大小为 0。
2. BlockingQueue 是一个接口，需要使用其实现类，例如 ArrayBlockingQueue。ArrayBlockingQueue 在创建时需要指定容量，这是必须的。

解决方案

以下提供两种创建同步阻塞队列列表的方法：使用 Stream API 和使用 for 循环。

1. 使用 Stream API

Stream API 提供了一种简洁的方式来初始化列表。

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;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class SynchronizedQueueList {

    public static void main(String[] args) {
        final int queueCapacity = 10; // 每个队列的容量
        final int limit = 5;        // 队列的数量

        List listOfQueues = Stream
                .generate(() -> new ArrayBlockingQueue(queueCapacity))
                .limit(limit)
                .collect(Collectors.collectingAndThen(
                        Collectors.toList(),
                        Collections::synchronizedList
                ));

        System.out.println("Size of synchronized list: " + listOfQueues.size()); // 输出 5
    }
}

代码解释:

• Stream.generate(() -> new ArrayBlockingQueue(queueCapacity))：创建一个无限流，每次生成一个新的 ArrayBlockingQueue 实例，并指定其容量。
• .limit(limit)：限制流的大小为 limit，即队列的数量。
• .collect(Collectors.collectingAndThen(Collectors.toList(), Collections::synchronizedList))：首先将流收集到一个 List 中，然后使用 Collections.synchronizedList 方法将其转换为同步列表。

2. 使用 For 循环

传统的 for 循环方式更加直观。

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 SynchronizedQueueListForLoop {

    public static void main(String[] args) {
        final int capacity = 10; // 每个队列的容量
        final int limit = 5;     // 队列的数量

        List> listOfQueues = new ArrayList<>();
        for (int i = 0; i < limit; i++) {
            listOfQueues.add(new ArrayBlockingQueue<>(capacity));
        }

        List synchronizedListOfQueues = Collections.synchronizedList(listOfQueues);

        System.out.println("Size of synchronized list: " + synchronizedListOfQueues.size()); // 输出 5
    }
}

代码解释:

• 创建一个 ArrayList 实例。
• 使用 for 循环，循环 limit 次，每次创建一个 ArrayBlockingQueue 实例，并将其添加到列表中。
• 使用 Collections.synchronizedList 方法将列表转换为同步列表。

注意事项

• ArrayBlockingQueue 容量: ArrayBlockingQueue 必须在创建时指定容量，这是强制性的。
• 线程安全: Collections.synchronizedList 返回的是一个线程安全的列表，但是对列表中元素的访问（例如，对 BlockingQueue 的 put 和 take 操作）仍然需要进行适当的同步，因为 Collections.synchronizedList 仅仅保证了对 List 接口方法的线程安全，而无法保证 List 中元素自身的线程安全。
• 其他 BlockingQueue 实现: 除了 ArrayBlockingQueue，还有其他 BlockingQueue 的实现，例如 LinkedBlockingQueue。LinkedBlockingQueue 可以在创建时指定容量，也可以不指定容量，如果不指定容量，则容量为 Integer.MAX_VALUE。选择哪种实现取决于具体的应用场景。

总结

创建包含多个 BlockingQueue 实例的同步列表，需要注意 BlockingQueue 的容量初始化以及列表本身的线程安全。Stream API 和 for 循环都是可行的方案，选择哪种方案取决于个人偏好和代码风格。重要的是要理解背后的原理，并根据实际情况选择最合适的方案。记住，即使列表是同步的，对列表中元素的访问仍然需要谨慎处理，以避免并发问题。