如何创建 Java 并发队列，我们​​可以在单个调用中阻塞接收超过 1 个元素？

Question

背景：我需要以异步方式向WebSocket客户端发送许多小消息。消息通常在高峰期发送，因此在暂停一段时间后，我需要快速发送约 5000 条消息。所以问题是：

• 我不想在单线程中启动 5000 个异步
• 我不想串行循环“启动异步”-“等待完成”5000 次
• 我不想使用 5000 个线程，每个线程只有一个“启动异步”-“等待完成”

最好的方法是为每个线程分组约 20 个异步，所以我需要非常具体的队列：

• lot of 表示队列中的并发推送/轮询
• small-sized asynchronous 表示我想以捆绑方式轮询，例如 每个队列有 1 到 20 条消息take()（这样我就可以启动 1...20 次异步 I/O并等待单线程完成）
• 立即 表示我不想等到 20 条消息被轮询，只有当队列有很多消息时才应该使用 bundle-poll。应立即轮询并发送一条消息。

所以基本上：我需要像队列这样的结构，在单个阻塞调用中具有阻塞 take(1 到 X) 等待元素。伪代码：

[each of ~50 processing threads]:
messages = queue.blockingTake( max 10 or at least 1 if less than 10 available );
for each message: message.startAsync()
for each message: message.waitToComplete()
repeat

Answer 1

如果不是真的需要，我不会从头开始实现队列。如果您有兴趣，请提供一些想法：

Queue> 如果你只有 1 个线程在做这些提议。如果您有更多，则必须同步该集合。比如，一个提供者 peek()-s 进入队列，发现最后一个集合有太多元素，所以它创建一个新集合并提供它。

或

许多正在运行的线程，其中可运行对象从队列中一个一个地获取元素。

或

每个发送线程 1 个队列，如果您保留队列引用，则可以以循环方式向每个队列添加元素。

或

子类化您选择的 BlockingQueue，并使用重写的普通 take() 版本创建“Collection take(int i)”方法。