java - 带有 delaySubscription 的 rxjava onStart 事件

Question

Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        subscriber.onStart();
        subscriber.onNext(1);
        subscriber.onCompleted();
    }
}).delaySubscription(5, TimeUnit.SECONDS).subscribe(new Subscriber<Integer>() {
    @Override
    public void onCompleted() {
        Log.e("TAG", String.format("(%s) - onCompleted", System.currentTimeMillis()));
    }

    @Override
    public void onError(Throwable e) {
        Log.e("TAG", String.format("(%s) - onError", System.currentTimeMillis()), e);
    }

    @Override
    public void onNext(Integer integer) {
        Log.e("TAG", String.format("(%s) - onNext: %s", System.currentTimeMillis(), integer));
    }

    @Override
    public void onStart() {
        super.onStart();
        Log.e("TAG", String.format("(%s) - onStart", System.currentTimeMillis()));
    }
});

输出:

(1485004553817) - onStart
(1485004558818) - onNext: 1
(1485004558819) - onCompleted

为什么 onStart 事件不等待 delaySubscription 并很快调用？

我想知道何时调用调用方法

Answer 1

文档说 -

开始 -

This method is invoked when the Subscriber and Observable have been connected but the Observable has not yet begun to emit items or send notifications to the Subscriber.

延迟订阅:

Returns an Observable that delays the subscription to the source Observable by a given amount of time.

onNext 仅在订阅完成时调用。 onStart 在建立连接时被调用。因此，它根据定义按预期工作。

您可以尝试注释代码 subscriber.onStart();并再次执行相同的操作，注意 onStart 仍然在开始时被调用。故意执行并没有真正调用上述方法，因为这不是在我们创建的真实订阅者上执行的，而是在 delaySubscription(类型为 OnSubscribeDelaySubscription)的结果上执行的。

下面是一个 fragment ，它可能会帮助您实现您正在寻找的东西:

   public static void main(String[] args) throws UnsupportedEncodingException, IOException {
      Observable.timer(5, TimeUnit.SECONDS).flatMap(val -> {
          System.out.println("Initialize");
          return Observable.create(subscriber -> {
              System.out.println("onsubscribe");
              doMyAsyncStuff(subscriber);
          });
      }).subscribe(val -> System.out.println(val));

      Observable.timer(10, TimeUnit.SECONDS).toBlocking().first();
   }

我们初始化一个 timer ，一旦定时器被执行，我们在flatMap中执行一些任务这应该与您之前对 onStart 所做的相同.一旦执行该任务，我们就会发出 Observable它会发出您之前可以使用 onNext 消耗的所有元素电话。