blocks|key|121905|text|您可以查看异步队列实现。如果您使用的是Java8，那么cyclops-react+(我是该项目的开发人员)提供了一个async.Queue，它将允许您异步地(干净地)填充和使用队列。|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|121906|例如：|121907|Queue<String>+queue+=+QueueFactories.<String>unboundedQueue().build();|code-block|syntax|javascript|121908|或者简单地说(只要这是一个com.aol.simple.react.async.Queue)|121909|Queue<String>+queue+=+new+Queue<>();|121910|然后在一个单独的线程中：|121911|new+Thread(()+->+{
++++++++while+(true)+{
++++++++++++queue.add("New+message+"+%2B+System.currentTimeMillis());
++++++++}
++++}).start();|121912|回到您的主线程，您的原始代码现在应该可以按预期工作(无限迭代添加到队列中的消息并将其打印出来)|121913|queue.stream().forEach(e+->+System.out.println(e));|121914|队列和流可以在任何阶段通过以下方式关闭：|121915|queue.close();|121916|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/aol/cyclops-react^0|R|D|0|0|0|0|0|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|18|8|@]|9|@$A|19|B|1A|1|1B]]|C|$]]|$1|D|3|E|5|6|7|1C|8|@]|9|@]|C|$]]|$1|F|3|G|5|H|7|1D|8|@]|9|@]|C|$I|J]]|$1|K|3|L|5|6|7|1E|8|@]|9|@]|C|$]]|$1|M|3|N|5|H|7|1F|8|@]|9|@]|C|$I|J]]|$1|O|3|P|5|6|7|1G|8|@]|9|@]|C|$]]|$1|Q|3|R|5|H|7|1H|8|@]|9|@]|C|$I|J]]|$1|S|3|T|5|6|7|1I|8|@]|9|@]|C|$]]|$1|U|3|V|5|H|7|1J|8|@]|9|@]|C|$I|J]]|$1|W|3|X|5|6|7|1K|8|@]|9|@]|C|$]]|$1|Y|3|Z|5|H|7|1L|8|@]|9|@]|C|$I|J]]|$1|10|3|-4|5|6|7|1M|8|@]|9|@]|C|$]]]|11|$12|$5|13|14|15|C|$16|17]]]]

You could look at an async Queue implementation. If you have Java 8, then <a href="https://github.com/aol/cyclops-react" rel="noreferrer">cyclops-react</a>, I a developer on this project, provides an async.Queue that will allow you to both populate and consume from the Queue asyncrhonously (and cleanly).

e.g. 

<pre><code>Queue&lt;String&gt; queue = QueueFactories.&lt;String&gt;unboundedQueue().build();
</code></pre>

Or simply (as long as this is a com.aol.simple.react.async.Queue)

<pre><code>Queue&lt;String&gt; queue = new Queue&lt;&gt;();
</code></pre>

Then in a separate thread :

<pre><code>new Thread(() -&gt; {
 while (true) {
 queue.add("New message " + System.currentTimeMillis());
 }
 }).start();
</code></pre>

Back on your main thread, your original code should now work as expected (infinetely iterate over the messages being added to the queue and print them out)

<pre><code>queue.stream().forEach(e -&gt; System.out.println(e));
</code></pre>

The Queue and hence the Stream can be closed at any stage via -

<pre><code>queue.close();
</code></pre>

blocks|key|121963|text|另一种方法是构建自定义Spliterator。在我的例子中，我有一个阻塞队列，我想构建一个继续提取元素的流，直到块超时。拆分器类似于：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|121964|public+class+QueueSpliterator<T>+implements+Spliterator<T>+{
++++private+final+BlockingQueue<T>+queue;
++++private+final+long+timeoutMs;

++++public+QueueSpliterator(final+BlockingQueue<T>+queue,+final+long+timeoutMs)+{
++++++++this.queue+=+queue;
++++++++this.timeoutMs+=+timeoutMs;
++++}

++++@Override
++++public+int+characteristics()+{
++++++++return+Spliterator.CONCURRENT+%7C+Spliterator.NONNULL+%7C+Spliterator.ORDERED;
++++}

++++@Override
++++public+long+estimateSize()+{
++++++++return+Long.MAX_VALUE;
++++}

++++@Override
++++public+boolean+tryAdvance(final+Consumer<?+super+T>+action)+{
++++++++try+{
++++++++++++final+T+next+=+this.queue.poll(this.timeoutMs,+TimeUnit.MILLISECONDS);
++++++++++++if+(next+==+null)+{
++++++++++++++++return+false;
++++++++++++}
++++++++++++action.accept(next);
++++++++++++return+true;
++++++++}+catch+(final+InterruptedException+e)+{
++++++++++++throw+new+SupplierErrorException("interrupted",+e);
++++++++}
++++}

++++@Override
++++public+Spliterator<T>+trySplit()+{
++++++++return+null;
++++}

}|code-block|syntax|javascript|121965|处理InterruptedException时抛出的异常是RuntimeException的一个扩展。使用这个类，可以通过:StreamSupport.stream(新的QueueSpliterator(...))构建一个流并添加通常的流操作。|121966|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|K|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|L|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|M|8|@]|9|@]|A|$]]|$1|I|3|-4|5|6|7|N|8|@]|9|@]|A|$]]]|J|$]]

Another approach is to build a custom Spliterator. In my case, I've got a blocking queue, and I want to build a stream that continues to extract elements, until the block times out. The spliterator is something like:

<pre><code>public class QueueSpliterator&lt;T&gt; implements Spliterator&lt;T&gt; {
 private final BlockingQueue&lt;T&gt; queue;
 private final long timeoutMs;

 public QueueSpliterator(final BlockingQueue&lt;T&gt; queue, final long timeoutMs) {
 this.queue = queue;
 this.timeoutMs = timeoutMs;
 }

 @Override
 public int characteristics() {
 return Spliterator.CONCURRENT | Spliterator.NONNULL | Spliterator.ORDERED;
 }

 @Override
 public long estimateSize() {
 return Long.MAX_VALUE;
 }

 @Override
 public boolean tryAdvance(final Consumer&lt;? super T&gt; action) {
 try {
 final T next = this.queue.poll(this.timeoutMs, TimeUnit.MILLISECONDS);
 if (next == null) {
 return false;
 }
 action.accept(next);
 return true;
 } catch (final InterruptedException e) {
 throw new SupplierErrorException("interrupted", e);
 }
 }

 @Override
 public Spliterator&lt;T&gt; trySplit() {
 return null;
 }

}
</code></pre>

The exception thrown to handle InterruptedException is an extension of RuntimeException. Using this class, one can build a stream via:
 StreamSupport.stream(new QueueSpliterator(...))
and add the usual stream operations.

Let's say we have a Queue

<pre><code>BlockingQueue&lt;String&gt; queue= new LinkedBlockingQueue&lt;&gt;();
</code></pre>

and some other thread puts values in it, then we read it like

<pre><code>while (true) {
 String next = queue.take();
 System.out.println("next message:" + next);
}
</code></pre>

How can I iterate over this queue in stream style, while maintaining similar semantics to above code.

This code only traverses the current queue state:

<pre><code>queue.stream().forEach(e -&gt; System.out.println(e));
</code></pre>

Stream API and Queues: Subscribe to BlockingQueue stream-style

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

假设我们有一个队列BlockingQueue<String> queue= new LinkedBlockingQueue<>();然后其他线程将值放入其中，然后我们将其读作while (true) { String next = queue.take(); System.out.println("next...

问流接口和队列:订阅BlockingQueue流样式
EN

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问流接口和队列:订阅BlockingQueue流样式EN

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问流接口和队列:订阅BlockingQueue流样式
EN