如何在 RxJava 中创建以固定间隔运行的作业？

Question

我正在尝试创建一个任务，该任务将定期查询我的数据库，将所有结果写入其他状态，我想使用 RxJava 来执行此操作。

我正在使用RxJava-JDBC 来查询我的数据库。代码如下所示：

    final Database db = Database.from(url);

    db
        .select("SELECT f1,f2 FROM mydata")
        .autoMap(MyDatum.class)
        .subscribe(
            new Action1<MyDatum>() {
                @Override
                public void call(MyDatum t) {
                    state.add(t);
                }
            },
            new Action1<Throwable>() {
                @Override
                public void call(Throwable t) {
                    L.error("Task failed", t);
                }
            },
            new Action0() {
                @Override
                public void call() {
                    state.makeAvailable();
                }
            }
        );

问题是，当我订阅时它会起作用，然后它就会停止。所以我使用了Observable.interval 并且有这个工作：

    Observable
        .interval(10, TimeUnit.SECONDS)
        .forEach(
            new Action1<Long>() {
                @Override
                public void call(Long arg) {
                    db
                        .select("SELECT f1,f2 FROM mydata")
                        .autoMap(MyDatum.class)
                        .subscribe(
                            new Action1<MyDatum>() {
                                @Override
                                public void call(MyDatum t) {
                                    state.add(t);
                                }
                            },
                            new Action1<Throwable>() {
                                @Override
                                public void call(Throwable t) {
                                    L.error("Task failed", t);
                                }
                            },
                            new Action0() {
                                @Override
                                public void call() {
                                    state.makeAvailable();
                                }
                            }
                        );
                }
            }
        );

但我想知道如果我没有做错什么让一个流嵌套在另一个流中。 我考虑过使用flatMap，但是onComplete 永远不会被执行，因为interval 永远不会调用onComplete。

我希望它不仅可以通过间隔触发，还可以通过传入事件触发。

我在这里遗漏了什么吗？ 谢谢

Answer 1

doOnNext 和 doOnCompleted 运算符在您的情况下非常有用。下面是一个示例，您可以如何使用这些运算符来实现所描述的行为：

final Observable<MyDatum> observable = Observable.interval(10, TimeUnit.SECONDS).flatMap(new Func1<Long, Observable<MyDatum>>() {
    @Override
    public Observable<MyDatum> call(final Long counter) {
        return db.select("SELECT f1,f2 FROM mydata")
                .autoMap(MyDatum.class)
                .doOnNext(new Action1<MyDatum>() {
                    @Override
                    public void call(final MyDatum value) {
                        state.add(value);
                    }
                })
                .doOnCompleted(new Action0() {
                    @Override
                    public void call() {
                        state.makeAvailable();
                    }
                });
    }
});

final Subscription subscription = observable.subscribe();

Answer 2

flatMap 和 merge 是您要使用的运算符。首先，您应该避免订阅操作员体内的可观察对象。而是使用 flatMap 并返回 observable。这将为您订阅所有发出的 observables。

为了手动触发查询，您可以合并 PublishSubject (documentation)，您可以调用 onNext 来推送事件并手动触发查询。把你的代码改成这样。

PublishSubject<Long> subject = PublishSubject.create();
Observable.merge(subject, Observable.timer(0, 1, TimeUnit.SECONDS))
    .flatMap(new Func1<Long, Observable<MyDatum>>() {
        @Override
        public Observable<MyDatum> call(Long arg) {
            return db
                .select("SELECT f1,f2 FROM mydata")
                .autoMap(MyDatum.class);
    }}).subscribe(
        new Action1<MyDatum>() {
            @Override
            public void call(MyDatum t) {
                state.add(t);
            }
        },
        new Action1<Throwable>() {
            @Override
            public void call(Throwable t) {
                L.error("Task failed", t);
            }
        },
        new Action0() {
            @Override
            public void call() {
                state.makeAvailable();
            }
        }
    );
// you can call onNext with any value to trigger a manual query
subject.onNext(999L);

这是一个演示此行为的简单 RxJava sn-p。

CountDownLatch l = new CountDownLatch(5);
PublishSubject<Long> subject = PublishSubject.create();
Observable.merge(subject, Observable.timer(0, 1, TimeUnit.SECONDS).take(3))
        .flatMap((Long arg) -> {
            System.out.println("tick: " + arg);
            l.countDown();
            return Observable.just(arg+10);
        })
        .forEach(System.out::println);
l.await(1, TimeUnit.SECONDS);
subject.onNext(999L);
l.await();

输出

tick: 0
10
tick: 999
1009
tick: 1
11
tick: 2
12