在主线程中运行 Flow答案

【问题标题】：Run Flow in main thread在主线程中运行 Flow
【发布时间】：2017-09-30 18:53:49
【问题描述】：

我再次将 RxJava 与 Java 9 Flow 进行比较。我看到 Flow 默认情况下是异步的，我想知道是否有办法让它同步运行。

有时我们只是想把它不是用于 Nio 而是用于糖语法，并拥有更同质的代码。

在 RxJava 中，默认情况下它是同步的，您可以在管道中使用 observerOn 和 subscribeOn 使其异步运行。

Flow 中有什么操作符可以让它在主线程中运行？

问候。

【问题讨论】：

标签： java java-9 java-flow

【解决方案1】：

您可以按照 Flow 中的说明定义自定义 Publisher 以使用同步执行。

一个非常简单的发布者，仅发布（在请求时）单个 TRUE 项目给单个订阅者。因为订阅者只收到单个项目，此类不使用缓冲和排序控制。

class OneShotPublisher implements Publisher<Boolean> {
   private final ExecutorService executor = ForkJoinPool.commonPool(); // daemon-based
   private boolean subscribed; // true after first subscribe
   public synchronized void subscribe(Subscriber<? super Boolean> subscriber) {
     if (subscribed)
       subscriber.onError(new IllegalStateException()); // only one allowed
     else {
       subscribed = true;
       subscriber.onSubscribe(new OneShotSubscription(subscriber, executor));
     }
   }
   static class OneShotSubscription implements Subscription {
     private final Subscriber<? super Boolean> subscriber;
     private final ExecutorService executor;
     private Future<?> future; // to allow cancellation
     private boolean completed;
     OneShotSubscription(Subscriber<? super Boolean> subscriber,
                         ExecutorService executor) {
       this.subscriber = subscriber;
       this.executor = executor;
     }
     public synchronized void request(long n) {
       if (n != 0 && !completed) {
         completed = true;
         if (n < 0) {
           IllegalArgumentException ex = new IllegalArgumentException();
           executor.execute(() -> subscriber.onError(ex));
         } else {
           future = executor.submit(() -> {
             subscriber.onNext(Boolean.TRUE);
             subscriber.onComplete();
           });
         }
       }
     }
     public synchronized void cancel() {
       completed = true;
       if (future != null) future.cancel(false);
     }
   }
 }

【讨论】：

【解决方案2】：

没有操作员可以这样做，但 API 允许您控制项目的发布方式。因此，您可以直接从当前线程调用订阅者方法。

class SynchronousPublisher implements Publisher<Data>  {
      public synchronized void subscribe(Subscriber<? super Data> subscriber) {
           subscriber.onSubscribe(new SynchronousSubscription(subscriber));
      }
 }
 static class SynchronousSubscription implements Subscription {
      private final Subscriber<? super Data> subscriber;

       SynchronousSubscription(Subscriber<? super Data> subscriber) {
          this.subscriber = subscriber;
       }
       public synchronized void request(long n) {
           ... // prepare item            
           subscriber.onNext(someItem);      
       }

       ...
   }
}

【讨论】：

谢谢，这也是一个很好的解决方法，仍然很冗长，但是嘿！这是Java :)
@paul 请记住，Java 将此作为 RS 规范（无实现）提供。它不是 RxJava 的替代品。仅作为规范所需的最低要求。

【解决方案3】：

这取决于您在主线程上运行的意思。

如果您想强制任意 Flow 在特定线程上执行，则没有标准方法可以执行此操作，除非 Flow 在库中实现，让您覆盖提供异步的部分。在 RxJava 术语中，这些是由 Schedulers 实用程序类提供的 Schedulers。

如果您想在主线程上观察 Flow，您必须在 Flow.Subscriber 之上编写一个阻塞队列消费者，它会阻塞线程，直到队列有项目。这可能会变得复杂，所以我将向您推荐 Reactive4JavaFlow 中的 blockingSubscribe 实现。

如果你想将Java主线程用作Executor/Scheduler，那就更复杂了，需要类似的阻塞机制以及线程池执行器的一些想法。 Reactive4JavaFlow 恰好有这样一个 Scheduler，您可以通过 new SubmissionPublisher<>(128, blockingScheduler::schedule) 将其用作 Executor。

【讨论】：