为什么这个 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. Observable 的生存时间规则是什么？是否有任何最佳实践应该如何引用运行时间更长的 Observable 以及它们是否可以是静态的？在我的测试中，我注意到这并不重要，但也许在这里，当涉及计时器时。

--

更正的代码[尚未工作]：

• 添加了重复（）

  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());

我是否需要从函数外部引用 Observable 才能使其存活更长时间？它是由 GC (Android) 收集的吗？函数是静态的有问题吗？

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

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

你是对的。没关系。

Answer 2

关于您的评论，为简单起见，您可以这样做，

Observable.timer(1500, 1500, TimeUnit.MILLISECONDS)
        .flatMap(new Func1<Long, Observable<Object>>() {
            @Override
            public Observable<Object> call(Long aLong) {
                String o = "0";
                return Observable.from(o);
            }
        })
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object aLong) {
                System.out.println(aLong);
            }
        });

在这里，您仍然可以获得计时器的好处，而无需在顶部添加拉链/重复。它仍然有点冗长，但它有点简单。