Java 8 Операции с итерационными потоками

Question

Я хочу выполнить поток, в котором вывод из потока будет использоваться в качестве источника для того же потока в той же операции.

В настоящее время я выполняю этот вид операции с использованием очереди; Я удаляю элемент, обрабатываю его и добавляю любые результаты, которые требуют дальнейшей обработки в очередь. Вот два примера такого рода вещи:

Queue<WorkItem> workQueue = new Queue<>(workToDo); 
while(!workQueue.isEmpty()){ 
    WorkItem item = workQueue.remove(); 
    item.doOneWorkUnit(); 
    if(!item.isDone()) workQueue.add(item); 
} 

Queue<Node> nodes = new Queue<>(rootNodes); 
while(!nodesLeft.isEmpty()){ 
    Node node = nodes.remove(); 
    process(node); 
    nodes.addAll(node.children()); 
} 

Я бы себе представить, что первые могут быть выполнены одновременно, как это:

try { 
    LinkedBlockingQueue<WorkItem> workQueue = new LinkedBlockingQueue<>(); 
    Stream<WorkItem> reprocess = Stream.generate(() -> workQueue.remove()).parallel(); 

    Stream.concat(workToDo.parallelstream(), reprocess) 
      .filter(item -> {item.doOneWorkUnit(); return !item.isDone();}) 
      .collect(Collectors.toCollection(() -> workQueue)); 
} catch (NoSuchElementException e){} 

И второе, как:

try { 
    LinkedBlockingQueue<Node> reprocessQueue = new LinkedBlockingQueue<>(); 
    Stream<WorkItem> reprocess = Stream.generate(() -> nodes.remove()).parallel(); 

    Stream.concat(rootNodes.parallelStream(), reprocess) 
      .filter(item -> {process(item); return true;}) 
      .flatMap(node -> node.children().parallelStream()) 
      .collect(Collectors.toCollection(() -> reprocessQueue)); 
} catch (NoSuchElementException e){} 

Однако , они чувствуют себя как неудобные обходные пути, и мне не нравится прибегать к использованию исключений. У кого-нибудь есть лучший способ сделать такие вещи?

Answer 1

Чтобы сделать работу параллельной, я бы использовал стандарт java.util.concurrent.Executor. Чтобы вернуть задачу в рабочую очередь, в конце кода каждой задачи добавьте executor.execute(this).