- iOS/Objective-C 元类和类别
- objective-c - -1001 错误,当 NSURLSession 通过 httpproxy 和/etc/hosts
- java - 使用网络类获取 url 地址
- ios - 推送通知中不播放声音
是否可以将传统的事件监听器重构为 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()
返回 void,我想保持这种状态,但我也愿意接受各种选择...
阅读 CompletableFutures 后Tomasz Nurkiewicz 对于类似的场景,存在一个替代选项,假设对客户端进行了微小的更改:
List<CompletableFuture<TaskResult>> taskFutures =
client.submitJobAsync( new MultiTaskJob())
在这里,客户端得到一个列表 CompletableFutures<TaskResult>
,因此我们需要在 future 完成时收集它们的结果:
//processes all task result futures
List<TaskResult> = taskFutures.stream()
.map(taskResult ->
taskResult.thenApply(this::processResult))
.collect(Collectors.<TaskResult>toList());
文章还说明了如何使用 CompletableFuture.allOf(..)
执行最终处理,但只有在所有 futures 完成之后(这很巧妙);在我的案例中,这就是聚合发生的地方。这里没有代码显示,尽管这篇文章很好地解释了它(我完全是流的 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 生成也将无休止地等待。
关于java - 使用事件监听器作为 Java 8 Stream 源,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/27670421/
当单击复选框(或选择所有复选框)时,我想向 CheckboxSelectionModel 添加一个监听器。 var smSensors = new xg.CheckboxSelectionModel(
我有一个简单的程序,允许 2 个客户端连接到服务器。 连接后,他们可以轮流点击空白卡片图像。 一旦 2 个客户中的任何一个点击空白卡片图片,卡片图片将变为 Ace 俱乐部图片。 更改将显示在客户端的两
我在这里看到了一个代码,该代码以字符串的形式检索鼠标的当前图标,但是此代码使用了TTimer来实现。 因此,我想知道是否存在某些事件(侦听器)来检测鼠标光标图标上的这些更改。 下面是使用TTimer的
我想在我的配置对象上获得一个 onload 事件。 以下工作,除非我创建一个 config.listeners={..} (我认为这就是我需要的?)替换 this.onload({...}); 我什至
通常,在 Java 中,当我有一个向其他对象提供某种通知的对象时,我将使用 Listener/Observer 模式。 有没有更类似于 Scala 的方式来做到这一点?我应该在 Scala 中使用这种
我有一个带有动画器的游戏对象和一些可以触发事件的动画(具有特定的回调函数)。 只要我将脚本添加到与动画器相同的游戏对象(包括接收器),一切都会正常工作: public class AnimatorEv
我有一个带有监听器的 DialogFragment,用于单击按钮以调用 fragment 中的函数。 我收到 lateinit property listener has not been initi
这个问题已经有答案了: Java ActionListener error: incompatible types (4 个回答) 已关闭 5 年前。 我最近刚刚开始学习 Java 代码。我在添加监听
我的代码遇到问题。我想知道是否有一种更简单的方法来使用监听器,而不是不断地这样做: example.addActionListener(new java.awt.event.ActionListene
有没有办法使用 .net 创建控制台应用程序。或通过某个端口监听 SMTP 消息的服务? 我需要创建一个中间层对象来捕获和处理 smtp 消息。也就是说,我希望该监听器发送和接收 smtp 消息,然后
我有一个带有动画器的游戏对象和一些可以触发事件的动画(具有特定的回调函数)。 只要我将脚本添加到与动画器相同的游戏对象(包括接收器),一切都会正常工作: public class AnimatorEv
我有许多向主事件生成服务注册的监听器。然而,我想告诉听众,事件流在某个时刻将会结束。您会通过简单地调用监听器上的方法(例如 finish())来完成此操作,还是有一个单独的事件方法 streamFin
我的代码有什么问题。 我创建了一个 JList,添加了项目并将其推到左侧(BorderLayout.WEST)。每次单击列表项时,我希望在列表右侧显示一个面板。但问题是,当选择列表项并运行监听器时,到
关闭。此题需要details or clarity 。目前不接受答案。 想要改进这个问题吗?通过 editing this post 添加详细信息并澄清问题. 已关闭 8 年前。 Improve th
这可能是一个简单的问题,但我没有看到它。 我有一个界面 public interface RenderableListener{ public void update(T element);
有人可以直接指出我的正确方向吗?当从组合框中选择适当的选项时,我希望小程序中的 Action 监听器显示从 html 文件检索的 jlabel 中的 3 个参数之一。 干杯 最佳答案 对于组合框,您需
我有一个网站,每个页面上都有许多 jQuery 事件处理程序,所有这些都在一个大型 .js 文件中。 这意味着对于任何页面,大多数事件处理程序都是针对不存在且根本不会使用的 HTML。 这会影响我的表
我有一些 jQuery 监听器设置,用于监听 type="text" 字段上的表单输入。但是,当用户从自动完成下拉框中选择一个选项(即他们之前输入的值已被记住以供将来使用)时,下面的监听器不会收集该值
关闭。这个问题需要多问focused 。目前不接受答案。 想要改进此问题吗?更新问题,使其仅关注一个问题 editing this post . 已关闭 9 年前。 Improve this ques
我正在用 unity (c#) 做一个类似国际象棋的游戏,但我在尝试进行向上转换以将信息从一个 child 发送到另一个抽象类时遇到了困难。 基本上,我有一个抽象类,它有一个带有函数的事件/委托(de
我是一名优秀的程序员,十分优秀!