如何将动态集合作为流处理？答案

【问题标题】：How to handle a dynamic collection as stream?如何将动态集合作为流处理？
【发布时间】：2017-04-04 14:50:47
【问题描述】：

Java 8 集合提供了将集合作为流获取的功能。但是，一旦我们调用 stream() 方法，我们就会将集合的当前内容作为流获取。如果我的集合在流处理期间增长怎么办？流上的操作可能会使用更多数据更新集合。有没有简单有效的方法来处理这种情况？

（我在流处理操作中尝试了 Stream.concat()，但出现异常：线程“main”java.lang.IllegalStateException 中的异常：流已被操作或关闭）

举个具体的例子，假设我有一个并发的url队列。

Queue<Url> concurrentUrlQue= initUrlQueue();

现在我想获取这个 url 队列的流，并一个一个地处理这些 URL。该过程包括从队列中删除 url、读取 url 指向的网页、从页面中提取 URL 并将这些 URL 添加到并发队列中。

concurrentUrlQue.stream().forEach((url)->readAndExtractUrls(url, concurrentUrlQue));

我希望能够将上述动态增长的队列作为流处理。（此外，我希望能够使用并行流处理这个动态队列）

有没有使用 java 流实现这一点的简单方法？

【问题讨论】：

您应该咨询javadoc of java.util.Spliterator。很长，但回答了您的问题。
我会通过它，但你能提示我拆分器如何解决我试图解决的问题吗？
所以你有一个 URL 可能会产生额外的 URL，可能会产生更多，你希望它们都在同一个集合中？听起来您想要某种可以并行运行的递归。拆分器听起来很有趣。
是的，我有一个队列，我处理的元素，处理后的结果被添加回队列，以便进一步处理它们。我处理了这个，甚至在不使用流的情况下并行处理。我想要一个流方法。
如果你真的想用流做这个，你只是在写一个尴尬的Spliterator，实际上只是一个ForkJoinTask。您应该改用该类及其附属机构ForkJoinPool。

标签： java java-stream

【解决方案1】：

您需要编写阻止等待新元素的拆分器。

class QueueSpliterator<T> extends Spliterators.AbstractSpliterator<T> {

  private final BlockingQueue<T> queue;

  public QueueSpliterator(BlockingQueue<T> queue) {
    super(Long.MAX_VALUE, 0);
    this.queue = queue;
  }

  public boolean tryAdvance(Consumer<? super T> action) {
    try {
      T element = queue.take();
      action.accept(element);
      return true;
    } catch (InterruptedException e) {
      return false;
    }
  }
}

然后您使用该拆分器创建一个流，并像处理普通的无限流一样处理它。

public class Main {
  public static void main(String... args) {
    BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(1000);

    new Thread(() -> {
      for (int i = 0; i < 1000; ++i) {
        try {
          queue.put(i);
        } catch (InterruptedException e) {
          throw new RuntimeException(e);
        }
      }
    }).start();


    Spliterator<Integer> queueSpliterator = new QueueSpliterator<>(queue);
    Stream<Integer> stream = StreamSupport.stream(queueSpliterator, false);

    stream.forEach(System.out::println);
  }
}

【讨论】：