使用事件监听器作为 Java 8 Stream 源

Question

快速-n-脏

是否可以将传统的事件监听器重构为 Java 8 Stream，使监听器事件成为流源？

长篇大论

客户端提交任意作业，然后监听结果：

Client client = new JobClient()
client.addTaskListener(this)
client.submitJobAsync( new MultiTaskJob())  //returns void, important (see below)

public void onTaskResult(TaskResult result){
  if(result.isLastResult())
    aggregateJobResults(result)
  else
    processResult(result)
}

问题

对于任何提交的作业，客户端会收到 n 个结果，但它不知道会收到多少个结果（它使用isLastResult() 来确定何时停止和聚合）。 p>

目标

我想将监听器重构为“供应商”或类似的东西，例如 onTaskResult() 是流源：

Supplier<TaskResult> taskResultSupplier = 
    () -> Stream.of( .. )   //onTaskResult() feeds this
            .map(result -> {
               if(result.isLastResult())
                 //logic here
             });

类似的东西；如果我能在客户不知道预期有多少结果的情况下做到这一点，那我就是黄金；现在，submitJobAsync() 返回 void，我想保持这种状态，但我也愿意选择......

替代方案

在阅读 Tomasz Nurkiewicz 的 CompletableFutures 以了解类似情况后，假设对客户端进行了细微更改，则存在替代选项：

List<CompletableFuture<TaskResult>> taskFutures = 
  client.submitJobAsync( new MultiTaskJob())

这里，客户端得到了CompletableFutures<TaskResult>的列表，所以我们需要收集futures完成时的结果：

//processes all task result futures 
List<TaskResult> = taskFutures.stream()
  .map(taskResult ->
     taskResult.thenApply(this::processResult))
  .collect(Collectors.<TaskResult>toList());

文章还说明了使用 CompletableFuture.allOf(..) 执行最终处理，但仅在 所有 期货完成之后（这很漂亮）；这就是我的情况发生聚合的地方。这里没有代码可以显示，虽然这篇文章很好地解释了它（我是一个有流的总 n00b，但如果我让它工作，我会发布代码：-D）

Answer 1

可以围绕您的 TaskResults 构建一个 Stream。看这个例子：

import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * Created for http://stackoverflow.com/q/27670421/1266906.
 */
public class AsyncToStream {
    public static void main(String[] args) {

        System.out.println("Unbuffered Test:");
        AsyncTaskResultIterator<TaskResult> taskListener1 = new AsyncTaskResultIterator<>();
        new TaskResultGenerator(taskListener1, 5).start();
        taskListener1.unbufferedStream().forEach(System.out::println);

        System.out.println("Buffered Test:");
        AsyncTaskResultIterator<TaskResult> taskListener2 = new AsyncTaskResultIterator<>();
        new TaskResultGenerator(taskListener2, 5).start();
        taskListener2.bufferedStream().forEach(System.out::println);
    }

    /**
     * This class wraps a sequence of TaskResults into an iterator upto the first TaskResult where {@code }isLastResult()} returns {@code true}
     */
    public static class AsyncTaskResultIterator<T extends TaskResult> implements Iterator<T>, TaskListener<T> {

        /**
         * This acts as an asynchronous buffer so we can easily wait for the next TaskResult
         */
        private final BlockingQueue<T> blockingQueue;
        /**
         * Becomes {@code true} once {@code TaskResult.isLastResult()} is received
         */
        private       boolean          ended;

        public AsyncTaskResultIterator() {
            blockingQueue = new LinkedBlockingQueue<>();
        }

        /**
         * Waits on a new TaskResult and returns it as long as the previous TaskResult did not specify {@code isLastResult()}. Afterwards no more elements can be retrieved.
         */
        @Override
        public T next() {
            if (ended) {
                throw new NoSuchElementException();
            } else {
                try {
                    T next = blockingQueue.take();
                    ended = next.isLastResult();
                    return next;
                } catch (InterruptedException e) {
                    throw new IllegalStateException("Could not retrieve next value", e);
                }
            }
        }

        @Override
        public boolean hasNext() {
            return !ended;
        }

        /**
         * Enqueue another TaskResult for retrieval
         */
        @Override
        public void onTaskResult(T result) {
            if (ended) {
                throw new IllegalStateException("Already received a TaskResult with isLastResult() == true");
            }
            try {
                blockingQueue.put(result);
            } catch (InterruptedException e) {
                throw new IllegalStateException("Could not enqueue next value", e);
            }
        }

        /**
         * Builds a Stream that acts upon the results just when they become available
         */
        public Stream<T> unbufferedStream() {
            Spliterator<T> spliterator = Spliterators.spliteratorUnknownSize(this, 0);
            return StreamSupport.stream(spliterator, false);
        }

        /**
         * Buffers all results and builds a Stream around the results
         */
        public Stream<T> bufferedStream() {
            Stream.Builder<T> builder = Stream.builder();
            this.forEachRemaining(builder);
            return builder.build();
        }
    }

    public static class TaskResultImpl implements TaskResult {
        private boolean lastResult;
        private String  name;

        public TaskResultImpl(boolean lastResult, String name) {
            this.lastResult = lastResult;
            this.name = name;
        }

        @Override
        public String toString() {
            return "TaskResultImpl{" +
                    "lastResult=" + lastResult +
                    ", name='" + name + '\'' +
                    '}';
        }

        @Override
        public boolean isLastResult() {
            return lastResult;
        }
    }

    public static interface TaskListener<T extends TaskResult> {
        public void onTaskResult(T result);
    }

    public static interface TaskResult {
        boolean isLastResult();
    }

    private static class TaskResultGenerator extends Thread {
        private final TaskListener<TaskResult> taskListener;
        private final int                      count;

        public TaskResultGenerator(TaskListener<TaskResult> taskListener, int count) {
            this.taskListener = taskListener;
            this.count = count;
        }

        @Override
        public void run() {
            try {
                for (int i = 1; i < count; i++) {
                    Thread.sleep(200);
                    taskListener.onTaskResult(new TaskResultImpl(false, String.valueOf(i)));
                }
                Thread.sleep(200);
                taskListener.onTaskResult(new TaskResultImpl(true, String.valueOf(count)));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

您没有提供您的 TaskResult 和 TaskListener 定义，所以我自己做了。 AsyncTaskResultIterator 仅适用于单个 TaskResult 序列。如果没有提供isLastResult() == true 的TaskResult next()，因此无缓冲流和缓冲流的生成也将无休止地等待。

Answer 2

正如 Brian Goetz 所说，为什么不试试 RxJava？

看看 takeWhile (https://github.com/ReactiveX/RxJava/wiki/Conditional-and-Boolean-Operators#takewhile) 和 map (https://github.com/ReactiveX/RxJava/wiki/Transforming-Observables#map)。