【发布时间】：2019-02-12 10:35:37
【问题描述】：
我有一个基于时间的算法，在某些情况下，不同的 observables 可能在同一时刻发生两次发射。
请看下面的代码：
int moment = ...; //calculated
Observable.timer(moment, unit)
.takeUntil(messages)
.map(e -> new Action())
messages Observable 像计时器一样在Schedulers.computation() 上执行。
当消息与计时器同时出现时，行为是随机的。有时会发出新的Action，有时不会，takeUntil(messages) 优先。
我需要完全控制这种行为。当然我可以在moment 上加几毫秒，就像这样它总是在messages 之后执行，但这不是我想要的。
如何告诉 RxJava messages observable 优先于计时器？
更新：
为了澄清这个问题，我添加了这个测试，每次执行都会成功
//before launching the test
RxJavaPlugins.setComputationSchedulerHandler(scheduler ->
new TestScheduler());
@Test
public void rx_precedence() {
int moment = 5;
Observable<Long> messages = Observable.timer(moment, TimeUnit.SECONDS);
TestObserver<Long> testObserver = Observable.timer(moment, TimeUnit.SECONDS)
.takeUntil(messages)
.test();
scheduler.advanceTimeTo(moment, TimeUnit.SECONDS);
testObserver.assertNoValues();
testObserver.assertTerminated();
}
在同一个moment 下，takeUntil 会被执行并且不会发出任何值。
但在服务器上，在现实世界中，当 je JVM 同时执行多项操作时，timer 可能优先于 takeUntil。
我需要一种方法来使这个优先级在任何环境中都保持稳定，这样当messages 在timer 用完的同时发出一个值时，就不会发出任何值。
【问题讨论】：
-
只是为了理解问题。您想确保在计时器用完时执行操作，除非
messages发出或完成，此时您什么都不做？ -
@Fred，我用测试更新了问题，你认为现在清楚了吗？
-
为什么你不只是尝试观察动作，而 observable 会在消息到来时调用？
-
如果依赖时间，可以观察消息并计算时间。当可观察的电话和消息收到时，您可以返回花费的时间
-
从测试看来问题可以通过改变调度器来解决。为什么不使用接收调度程序的
timer的变体？将它们设置为trampoline并查看它是否有效。我不确定这是否是一个合适的解决方案，但它可能会有所帮助。