ダイナミックコレクションをストリームとして扱うにはどうすればいいですか？

Question

Java 8コレクションは、コレクションをストリームとして取得する機能を提供します。しかし、いったんstream（）メソッドを呼び出すと、コレクションの現在の内容をストリームとして取得します。 ストリーム処理中にコレクションが大きくなった場合はどうなりますか？ストリーム上の操作によって、コレクションがより多くのデータで更新される可能性があります。この状況に対処するには、単純な&効果的な方法がありますか？

（Iストリーム処理操作の中から）（Stream.concatを試みたが、私は例外を取得：スレッドの例外「メイン」java.lang.IllegalStateException：ストリームが既にオペレートまたは閉鎖された）

は、撮影具体的な例として、URLの並行キューがあるとします。

Queue<Url> concurrentUrlQue= initUrlQueue(); 

ここで、このURLキューのストリームを取得し、そのURLを1つずつ処理します。このプロセスでは、キューからURLを削除し、URLが指すWebページを読み込み、ページからURLを抽出し、それらのURLを並行キューに追加します。

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

私は上記の動的に成長するキューをストリームとして処理できるようにします。 （さらに、この動的キューを並列ストリームで処理できるようにしたい）

Javaストリームを使用してこれを実現する簡単な方法はありますか？

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