android - 为什么这个 observable 只发出一个值

Question

我有以下代码基于@a.bertucci 此处提供的示例 Emit objects for drawing in the UI in a regular interval using RxJava on Android ，我在这里压缩了一个带定时器的 Observable。当我通过调用 processDelayedItems() 触发订阅时，压缩 Observable 中的代码 [A] 只执行一次，并且一个项目被发送到 [B]。我原以为代码 [A] 会在触发后连续运行并每 1500 毫秒持续发射一次元素，但显然它在这里只运行一次。

private static void processDelayedItems() {

    Observable.zip(
            Observable.create(new Observable.OnSubscribe<Object>() {

                @Override public void call(Subscriber<? super Object> subscriber) {
                    // [A] this code is only called once
                    subscriber.OnNext(o)
                }

            }),
            Observable.timer(1500, 1500, TimeUnit.MILLISECONDS), new Func2<Object, Long, Object>() {
                @Override public Object call(Object entity, Long aLong) {
                    return entity;
                }
            }
    )
    .subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<Object>() {

                @Override public void call(Object entity) {
                    // ... and accordingly one item is emitted [B]
                }

            }, new Action1<Throwable>() {

                @Override public void call(Throwable throwable) {
                    throwable.printStackTrace();
                }

            }, new Action0() {

                @Override public void call() {

                }

            });

}

1. 有人能看到我这里遇到的问题吗？是不是我需要从函数外部引用 Observable 以使其存活更长时间？它是由GC(Android)收集的吗？函数是静态的有问题吗？

2. Observables 的生存时间规则是什么？是否有任何最佳实践应该如何引用运行时间更长的 Observables 以及它们是否可以是静态的？在我的测试中，我注意到这并不重要，但在这里，当涉及到计时器时，它可能很重要。

--

更正后的代码[尚未生效]:

• 添加了 repeat()

  Observable.zip(
          Observable.create(new Observable.OnSubscribe<Object>() {
  
              @Override public void call(Subscriber<? super Object> subscriber) {
                  // [A] this code is only called once
                  subscriber.OnNext(o);
                  subscriber.OnCompleted();
              }
  
          }).repeat(Schedulers.newThread()),
          Observable.timer(1500, 1500, TimeUnit.MILLISECONDS), new Func2<Object, Long, Object>() {
              @Override public Object call(Object entity, Long aLong) {
                  return entity;
              }
          }
  )
  .subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread())
          .subscribe(new Action1<Object>() {
  
              @Override public void call(Object entity) {
                  // ... and accordingly one item is emitted [B]
              }
  
          }, new Action1<Throwable>() {
  
              @Override public void call(Throwable throwable) {
                  throwable.printStackTrace();
              }
  
          }, new Action0() {
  
              @Override public void call() {
  
              }
  
          });
  

Answer 1

您需要repeat 来生成无限的 Observable。例如，

    Observable.create(new Observable.OnSubscribe<Object>() {

        @Override public void call(Subscriber<? super Object> subscriber) {
            // [A] this code is only called once
            if (!subscriber.isUnsubscribed()) {
                subscriber.onNext(o);
            }
            if (!subscriber.isUnsubscribed()) {
                subscriber.onCompleted();
            }
        }

    }).repeat(Schedulers.newThread());

Is it that I need to reference the Observable from outside the function to keep it alive for more time? Is it collected by GC (Android)? Is it a problem that the function is static?

由于您使用了 Schedulers.newThread() 和 timer，所以会有一些其他线程引用了您的 Observable。您不需要更多工作。

What are the rules for Observables in terms of their livetime? Are there any best practices how longer-running Observables should be referenced and if they can be static at all? In my tests I noticed that it doesn't really matter, but maybe it does here, when a timer is involved.

你是对的。没关系。