如何发出结果使用定时器可观察与平面图

Question

我以以下方式使用rxjava进行维护任务：

在需要定期维护的类中，我使用以下静态订阅，这会导致在将类加载到内存中时首次启动可观察，然后按照指定的时间间隔定期启动。

private static Subscription subscription = Observable.timer(0, 5, TimeUnit.SECONDS)
        .flatMap(new Func1<Long, Observable<String>>() {
            @Override public Observable<String> call(Long aLong) {

                // some code

                return Observable.just(null);
            }
        }).subscribeOn(Schedulers.newThread()).observeOn(Schedulers.newThread())
        .subscribe();

现在我遇到了一种情况，我想向UI报告我的维护结果。

通常，我会使用以下模式

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

        @Override public void call(Subscriber<? super String> subscriber) {

            // some code

            subscriber.onNext(result);
            subscriber.onCompleted();            }

    }).subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<String>() {

                @Override public void call(String result) {

                    // write to the UI

                }

            });

但是这里我们有一个可观察的，它只执行一次。

对于一个定期执行的可观察的，我找不到在订阅者中调用Action的方法，这样我就可以使用subscriber.onNext（）传递结果。看起来可观察的没有合适的签名，它可能会花费定时器（）的很长时间，同时允许订阅动作。但是知道rxjava，我肯定有一个技巧；-）

我可以使用zip来压缩Timer可观察和一次性可观察（基本上将两个版本压缩在一起），但我宁愿使用第一个结构，因为它的行为略有不同。

--

我尝试通过以下方式将两个版本合并为一个：

private static Subscription subscription = Observable.timer(0, 5, TimeUnit.SECONDS)
        .flatMap(new Func1<Long, Observable<String>>() {
            @Override public Observable<String> call(Long aLong) {

                // some code // stays here to ensure there is no concurrency while executing

                final String result = "result"; // I store the result in a final variable after some code has been finished

                return Observable.create(new Observable.OnSubscribe<String>() {
                    @Override public void call(Subscriber<? super String> subscriber) {
                        subscriber.onNext(result); // then I use it in a new Observable and emit it
                        subscriber.onCompleted(); // not sure if this is needed here (haven't tested this yet)
                    }
                });

            }
        }).subscribeOn(Schedulers.newThread()).observeOn(Schedulers.newThread())
        .subscribe(new Action1<String>() {
            @Override public void call(String result) {
                // so I can finally consume the result on the UI thread
            }
        });

我创建了一个允许我向订阅者发送结果的对象，而不是创建并发出一个“null”可观察对象。

很乱，但这应该行得通，对吧？有更简单的解决方案吗？你有什么想法？

Answer 1

我知道这是很久以前才给这个问题添加答案的。然而，它可能会帮助其他人。据我所知，您希望创建一个observable并运行它，例如每3秒运行一次。因此，请在下面找到我的示例（您可以发出一个现成的操作符，如Observable.just，甚至可以使用Observable.create操作符创建您自己的Observable）：

Observable.interval(3,TimeUnit.SECONDS).flatMap(new Function<Long, ObservableSource<String>>() {
        @Override
        public ObservableSource<String> apply(Long aLong) throws Exception {

            return Observable.just("Hello " + Thread.currentThread().getName()+aLong);
           /*
            return Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                    try{
                        String data = fetchData("http://www.yahoo.com");
                        emitter.onNext(data.length()+"");
                        emitter.onComplete();
                    }catch (Exception e){
                        emitter.onError(e);
                    }
                }
            });
            */
        }
    }).subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            Log.d("RxResult","--  " + Thread.currentThread().getName() + " -- " +s);
        }
    });

public String fetchData(String url) throws IOException {
    BufferedReader in = new BufferedReader(new InputStreamReader(new URL(url).openStream()));
    String inputLine;
    String result = "";

    while((inputLine = in.readLine()) != null){
        result += inputLine;
    }

    return result.length()+"";
}

Answer 2

我不是舒尔，无法理解您的问题。

您想从您的可观察计时器和其他可观察计时器使用相同的订阅吗？

也许你可以用一个主题作为你的观察对象之间的桥梁。

Subject<?, ?> bridge = PublishSubject.create();
Observable.timer(5, SECONDS).flatMap(/* whatever*/).mergeWith(bridge).subscribe(/* toDo */);
Observable.create(/* subscription */).subscribe(bridge);

你的“一次性可观测”发出的每一项都会被推到brige中，brige会将其推到“计时器”可观测中。

这就是你要找的吗？