在 RxJava 中创建同步间隔

Posted 2023-05-07

【中文标题】在 RxJava 中创建同步间隔

【英文标题】：Create synchronization interval in RxJava

【发布时间】：2020-07-03 09:34:48

【问题描述】：

我需要以 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 秒并再次重复该过程。

【问题讨论】：

我认为其中一些要求有点 oncpomatible 1.“我需要以 4,8,12,16... 秒的特定间隔运行 2 个 JOB，另一个是 5,9 ,13,17...秒。” 2.“当作业 A 运行时，作业 B 应该休眠，反之亦然” 3 “运行作业 A 4 秒，然后运行作业 B 4 秒，然后再次重复该过程。”如果两个作业都需要 4 秒，并且它们必须在另一个运行时休眠，那么它们就不能以您定义的开始时间间隔工作

让我澄清一下。假设我们有一个异步任务，我们从 onPostexecute 调用另一个异步任务。在第二个异步任务完成其操作后，它再次调用第一个异步任务并继续循环

【参考方案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。

【讨论】：

感谢 Nicolas 的回答。这会奏效。除此之外，我仍然会等待任何更好的方法:)

上述答案的问题是JobA会运行在4,8,12,16...而JobB会运行在5,9,13...但是JobB不依赖于JobA。两者将同时运行。

效果应该是一样的。除非 jobA 需要超过 1 秒，在这种情况下你想跳过 jobB？

是的。的种类。我需要使这个同步。也就是说，JobB 应该在 JobA 之后开始，但是每个线程的生命周期应该是 4 秒。

所以 JobA 在 JobB 启动时被取消，反之亦然？ JobA 的最大持续时间是 1 秒，而 JobB 是 3 秒？

【参考方案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 秒，以最晚发生的为准。

【讨论】：

感谢您的回答。 jobA 和 jobB 都不会发出任何数据，但会在功能块中执行一些操作。但想法是让它同步运行。 JobA 将在 jobB 启动后需要 4 秒才能完成，也需要 4 秒。 jobA 应该再次开始，迭代将继续....