raf*_*ian 8 java future java-8 java-stream
是否可以将传统事件侦听器重构为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()确定何时停止和聚合).
我想将侦听器重构为"供应商"或类似的东西,例如onTaskResult()流源:
Supplier<TaskResult> taskResultSupplier =
() -> Stream.of( .. ) //onTaskResult() feeds this
.map(result -> {
if(result.isLastResult())
//logic here
});
像这样的东西; 如果我能在没有客户知道会有多少结果的情况下做到这一点,我就是金色的; 现在,submitJobAsync()返回无效,我想保持这种方式,但我也愿意接受选择......
在阅读了Tomasz Nurkiewicz针对类似场景的CompletableFutures之后,假设对客户进行了微小的更改,则存在备用选项:
List<CompletableFuture<TaskResult>> taskFutures =
client.submitJobAsync( new MultiTaskJob())
在这里,客户获得一份清单CompletableFutures<TaskResult>,因此我们需要在完成时收集期货的结果:
//processes all task result futures
List<TaskResult> = taskFutures.stream()
.map(taskResult ->
taskResult.thenApply(this::processResult))
.collect(Collectors.<TaskResult>toList());
这篇文章还说明了CompletableFuture.allOf(..)用于执行最终处理但仅在所有期货完成之后(它非常光滑); 这就是我的情况下会发生聚合的地方.无代码显示,在这里,虽然文章做了伟大的工作,解释它(我与流的总的n00b,但如果我得到它的工作我会后的代码:-D)
可以围绕您的 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()，则无缓冲 Stream 和缓冲 Stream 的生成也将无休止地等待。