为什么这个可观察的只发出一个值

Question

下面的代码基于@a.bertucci提供的一个示例，这里使用Android上的RxJava以固定的间隔在UI中绘制对象，其中我使用计时器压缩了一个可观察对象。当我通过调用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() {

                }

            });

}

>

有人能看到我这里的问题吗？是不是我需要从函数外部引用可观察的来让它保持更长时间的活动？它是由GC（Android）收集的吗？函数是静态的是个问题吗？

就生存时间而言，可观察事物的规则是什么？是否有任何最佳实践可以引用多长时间的可观察对象，以及它们是否可以是静态的？在我的测试中，我注意到这并不重要，但在这里，当涉及计时器时，它可能会起作用。

--

更正代码[尚未工作]：

>

添加重复（）

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

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

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

在这里，您仍然可以享受计时器带来的好处，而无需在顶部添加zip/repeat。这仍然有点冗长，但更简单一点。

Answer 2

您需要重复才能生成无限的可观测值。例如。，

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

我是否需要从函数外部引用可观察到的对象，以使其在更长的时间内保持活动状态？是否由GC（Android）收集？函数是静态的，这是一个问题吗？

因为您使用了调度程序。newThread（）和计时器，将有一些其他线程引用您的可观察线程。你不需要更多的工作。

就生存时间而言，可观察事物的规则是什么？是否有任何最佳实践可以引用多长时间的可观察对象，以及它们是否可以是静态的？在我的测试中，我注意到这并不重要，但在这里，当涉及计时器时，它可能会起作用。

你说得对。没关系。