spring响应式编程系列：异步生产数据

示例

大致流程

create

new MonoCreate

new LambdaMonoSubscriber

success

onNext

时序图

类图

数据发布者

MonoCreate

数据订阅者

LambdaMonoSubscriber

订阅的消息体

DefaultMonoSink

本篇文章我们来研究如何将现有异步 API（如回调式接口）适配到 Reactor 的响应式流中。

默认情况下，Mono.create的代码块执行在订阅时的线程上，但如果在该代码块中启动其他线程或使用异步API，那么数据生产就会变成异步的。示例如下所示：

示例

Mono<String> mono = Mono.create(sink -> {
    // 模拟一个异步API操作
    new Thread(() -> {
        try {
            Thread.sleep(1000); // 模拟耗时操作
            log.info("success");
            sink.success("Hello, World!"); // 成功时发射数据
        } catch (InterruptedException e) {
            sink.error(e); // 发生错误时发射错误信号
        }
    }).start();
});
log.info("main start");
mono.subscribe(x -> log.info("main finish"));
Thread.sleep(5000);

在这里，通过Mono.create模拟一个异步API操作，API操作成功后，调用sink.success("Hello, World!")进行数据发布者发送数据，从而触发数据的订阅。

接下来，让我们一起看看程序的流程是怎么处理的。

点击create()方法，如下所示：

大致流程

create

public static <T> Mono<T> create(Consumer<MonoSink<T>> callback) {
return onAssembly(new MonoCreate<>(callback));
}

在这里，new一个MonoCreate对象并返回。

点击MonoCreate，如下所示：

new MonoCreate

final class MonoCreate<T> extends Mono<T> implements SourceProducer<T> {
   static final Disposable TERMINATED = OperatorDisposables.DISPOSED;
   static final Disposable CANCELLED = Disposables.disposed();
   final Consumer<MonoSink<T>> callback;
   MonoCreate(Consumer<MonoSink<T>> callback) {
      this.callback = callback;
   }

在这里，将create()方法的回调接口参数赋值给callback属性。因此，Mono.create的参数就作为数据发布者的一个属性信息了。

点击示例里的mono.subscribe()，如下所示：

public final Disposable subscribe(
      @Nullable Consumer<? super T> consumer,
      @Nullable Consumer<? super Throwable> errorConsumer,
      @Nullable Runnable completeConsumer,
      @Nullable Context initialContext) {
   return subscribeWith(new LambdaMonoSubscriber<>(consumer, errorConsumer,
         completeConsumer, null, initialContext));
}

在这里，new一个LambdaMonoSubscriber对象，如下所示：

new LambdaMonoSubscriber

LambdaMonoSubscriber(@Nullable Consumer<? super T> consumer,
      @Nullable Consumer<? super Throwable> errorConsumer,
      @Nullable Runnable completeConsumer,
      @Nullable Consumer<? super Subscription> subscriptionConsumer,
      @Nullable Context initialContext) {
   this.consumer = consumer;
   this.errorConsumer = errorConsumer;
   this.completeConsumer = completeConsumer;
   this.subscriptionConsumer = subscriptionConsumer;
   this.initialContext = initialContext == null ? Context.empty() : initialContext;
}

在这里，将subscribe的回调接口参数赋值给consumer 属性，因此，mono.subscribe的参数就作为数据消费者的属性了。

点击上一步的subscribeWith()方法，如下所示：

monoCreate.subscribe

@Override
public void subscribe(CoreSubscriber<? super T> actual) {
   DefaultMonoSink<T> emitter = new DefaultMonoSink<>(actual);
   actual.onSubscribe(emitter);
   try {
      callback.accept(emitter);
   }
   catch (Throwable ex) {
      emitter.error(Operators.onOperatorError(ex, actual.currentContext()));
   }
}

在这里，首先调用了数据消费者的onSubscribe()方法，这个与《spring响应式编程系列：总体流程》一样。

另外，调用了callback.accept()方法，也就是Mono.create()的回调接口参数。

accept

Mono<String> mono = Mono.create(sink -> {
    // 模拟一个异步操作
    new Thread(() -> {
        try {
            Thread.sleep(1000); // 模拟耗时操作
            log.info("success");
            sink.success("Hello, World!"); // 成功时发射数据
        } catch (InterruptedException e) {
            sink.error(e); // 发生错误时发射错误信号
        }
    }).start();
});

在这里，模拟了耗时操作，然后调用sink.success()方法。

通常，可以将sink对象保存在线程共享环境里，等其它的业务操作执行完成后，再调用sink.success()方法，即可发射数据发布者数据，从而触发消费者订阅。

点击sink.success()，如下所示：

success

public void success(@Nullable T value) {

... ...
     for (; ; ) {
      int s = state;
      if (s == HAS_REQUEST_HAS_VALUE || s == NO_REQUEST_HAS_VALUE) {
         Operators.onNextDropped(value, actual.currentContext());
         return;
      }
      if (s == HAS_REQUEST_NO_VALUE) {
         if (STATE.compareAndSet(this, s, HAS_REQUEST_HAS_VALUE)) {
            try {
               actual.onNext(value);
               actual.onComplete();
            }
            catch (Throwable t) {
               actual.onError(t);
            }
            finally {
               disposeResource(false);
            }
         } else {
            Operators.onNextDropped(value, actual.currentContext());
         }
         return;
      }
      ... ...
   }
}

在这里，调用了数据订阅者的onNext()方法，如下所示：

onNext

public final void onNext(T x) {
   Subscription s = S.getAndSet(this, Operators.cancelledSubscription());
   if (s == Operators.cancelledSubscription()) {
      Operators.onNextDropped(x, this.initialContext);
      return;
   }
   if (consumer != null) {
      try {
         consumer.accept(x);
      }
      catch (Throwable t) {
         Exceptions.throwIfFatal(t);
         s.cancel();
         doError(t);
      }
   }
   if (completeConsumer != null) {
      try {
         completeConsumer.run();
      }
      catch (Throwable t) {
         Operators.onErrorDropped(t, this.initialContext);
      }
   }
}

时序图

类关系的设计，与《spring响应式编程系列：总体流程》类似，主要包括数据发布者对象、数据订阅者对象及订阅的消息体对象；
Mono和MonoCreate是数据发布者，LambdaMonoSubscriber是数据订阅者，DefaultMonoSink是订阅的消息体；
不同点在于，DefaultMonoSink可以通过示例里的Mono.create暴露给业务侧，业务侧的相关业务执行完成之后，可以通过调用该对象success方法，来触发订阅者的回调函数。