流API和队列：订阅BlockingQueue流式 [英] Stream API and Queues: Subscribe to BlockingQueue stream-style

问题描述

假设我们有一个队列

BlockingQueue<String> queue= new LinkedBlockingQueue<>();

和其他一些线程将值放入其中，然后我们读它就像

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

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

如何以流方式迭代此队列，同时保持类似的语义以上代码。

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:

queue.stream().forEach(e -> System.out.println(e));

推荐答案

我猜你有点什么期待，但我认为我有一个很好的预感。

I'm guessing a bit at what you're expecting, but I think I have a good hunch.

队列流，如迭代队列，代表当前内容队列。当迭代器或流到达队列的尾部时，它不会阻止等待添加的其他元素。迭代器或流在那时耗尽，计算终止。

The stream of a queue, like iterating over a queue, represents the current contents of the queue. When the iterator or the stream reaches the tail of the queue, it doesn't block awaiting further elements to be added. The iterator or the stream is exhausted at that point and the computation terminates.

如果你想要一个由队列的所有当前和未来元素组成的流，你可以做类似这样的事情：

If you want a stream that consists of all current and future elements of the queue, you can do something like this:

Stream.generate(() -> {
        try {
            return queue.take();
        } catch (InterruptedException ie) {
            return "Interrupted!";
        }
    })
    .filter(s -> s.endsWith("x"))
    .forEach(System.out::println);   

（不幸的是需要处理 InterruptedException make这非常麻烦。）

(Unfortunately the need to handle InterruptedException makes this quite messy.)

请注意，无法关闭队列，并且无法使用 Stream.generate 停止生成元素，所以这实际上是一个无限的流。终止它的唯一方法是使用短路流操作，例如 findFirst 。

Note that there is no way to close a queue, and there is no way for Stream.generate to stop generating elements, so this is effectively an infinite stream. The only way to terminate it is with a short-circuiting stream operation such as findFirst.

这篇关于流API和队列：订阅BlockingQueue流式的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！