在 RxJava 中创建同步间隔

Question

我需要以 4,8,12,16...秒的特定间隔运行 2 JOB，另一个是 5,9,13,17...秒。 p>

我在 RxJava 中使用了区间运算符。作业 B 需要在作业 A 之后运行。作业 B 应该在作业 A 运行时休眠，反之亦然。到目前为止，代码如下所示

var compositeDisposable = CompositeDisposable()
compositeDisposable.add(Observable.interval(0, recordIntervalPeriod, TimeUnit.MILLISECONDS)
                        .serialize()
                        .subscribe {
                            JobA()
                        })
compositeDisposable.add(Observable.interval(0, recorderStopIntervalStartTime, TimeUnit.MILLISECONDS)
                        .serialize()
                        .subscribe {
                            JobB()
                        })

在关注方面需要帮助

1.使用 RxJava 实现上述目标的最佳方法

2。运行 JobA 4 秒，然后运行 ​​JobB 4 秒并再次重复该过程。

Answer 1

我建议您使用每秒运行的单个作业，并根据计数器值决定每次调用哪个作业：

val disposable = Observable.interval(1, TimeUnit.SECONDS)
        .serialize()
        .subscribe { counter ->
            if (counter % 4 == 0L) {
                jobA()
            } else if ((counter - 1) % 4 == 0L) {
                jobB()
            }
        }

如果你仍然想使用两个 observables，我认为这也可以：

val disposable = CompositeDisposable()
disposable.addAll(
        Observable.interval(4, TimeUnit.SECONDS)
                .subscribe {
                    jobA()
                },
        Observable.interval(4, TimeUnit.SECONDS)
                .delay(1, TimeUnit.SECONDS)
                .subscribe {
                    jobB()
                })

免责声明：我没有经常使用 RxJava。

Answer 2

怎么样

Observable.interval(4,TimeUnit.SECONDS)
    .flatMap({
        jobA().zipWith(Observable.timer(1, TimeUnit.SECONDS) }
            .flatMap { jobB() }
    }, maxConcurrent = 1).subscribe()

我假设 jobA() 和 jobB() 是某种可观察的。

作业 A 应该等待作业 B 完成，因为最大并发设置为 1。

作业 B 应等待作业 A 或作业 A 开始后 1 秒，以最晚发生的为准。