3

我一直在尝试不同的方法来处理结果断开连接的阻塞方法,同时保持可能已被中断的状态。我发现必须处理发送和接收难以对齐的不同类和方法是令人沮丧的。

在下面的示例中,SomeBlockingMethod()通常返回void作为消息发送到其他进程。但相反,我synchronized用一个接收结果的侦听器来实现它。通过将其旋转到一个线程,我可以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与完整的调度程序方法相比,使用的一个警告是,存在于外部的任何先前状态都必须是最终的或有效的最终状态。

4

1 回答 1

0

您应该查看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);)
于 2016-04-28T14:06:36.133 回答