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

Question

Java8集合提供了以流形式获取集合的特性。然而，一旦我们调用stream（）方法，我们就会以stream的形式获得集合的当前内容。如果我的集合在流处理过程中增长怎么办？流上的操作可能会用更多的数据更新集合。有没有一个简单有效的方法来处理这种情况？

Queue<Url> concurrentUrlQue= initUrlQueue();

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

Answer 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);
  }
}