我一直在尝试不同的方法来处理具有断开连接结果的阻塞方法,同时维护可能已中断的状态。我发现必须处理发送和接收难以协调的不同类和方法是令人沮丧的。
在下面的示例中,当消息发送到其他进程时,SomeBlockingMethod()
通常返回 void
。但我却让它与接收结果的监听器同步
。通过将其分离到线程,我可以超时或无限期地 wait()
结果。
这很好,因为一旦返回结果,我就可以继续处理特定的状态,在等待线程任务的结果时我必须暂停该状态。
我的方法有什么问题吗?
尽管这个问题看起来很笼统,但我正在专门寻找有关 Java 中线程的建议。
示例伪代码:
public class SomeClass implements Command {
@Override
public void onCommand() {
Object stateObject = new SomeObjectWithState();
// Do things with stateObject
Runnable rasync = () -> {
Object r = SomeBlockingMethod();
// Blocking method timed out
if (r == null)
return;
Runnable rsync = () -> {
// Continue operation on r which must be done synchronously
// Also do things with stateObject
};
Scheduler().run(rsync);
};
Scheduler().run(rasync);
}
使用 CompletableFuture 更新:
CompletableFuture<Object> f = CompletableFuture.supplyAsync(() -> {
return SomeBlockingMethod();
});
f.thenRun(() -> { () -> {
String r = null;
try {
r = f.get();
}
catch (Exception e) {
e.printStackTrace();
}
// Continue but done asynchronously
});
或者更好:
CompletableFuture.supplyAsync(() -> {
return SomeBlockingMethod();
}).thenAccept((
Object r) -> {
// Continue but done asynchronously
});
严格使用CompletableFuture
的问题是,CompletableFuture.thenAccept
是从全局线程池运行的,并且不能保证与调用线程同步。
为同步任务添加调度程序可以修复此问题:
CompletableFuture.supplyAsync(() -> {
return SomeBlockingMethod();
}).thenAccept((
Object r) -> {
Runnable rsync = () -> {
// Continue operation on r which must be done synchronously
};
Scheduler().run(rsync);
});
与完整的调度程序方法相比,使用 CompletableFuture 的一个警告是,外部存在的任何先前状态都必须是最终的或实际上是最终的。
最佳答案
您应该查看RxJava ,它使用流操作并具有线程支持。
api.getPeople()
.observeOn(Schedulers.computation())
.filter(p -> return p.isEmployee();)
.map(p -> return String.format("%s %s - %s", p.firstName(), p.lastName(), p.payrollNumber());)
.toList()
.observerOn(<ui scheudler>)
.subscirbe(p -> screen.setEmployees(p);)
关于java - 委托(delegate)给线程,同时保持线性可读性,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/36916883/