当两个 Observables 同时变化时合并 Observables

Question

我正在尝试使用 RxPY 将 ReactiveX 集成到我的 GUI 中。这是一个通用的 ReactiveX 问题。

假设我有一个依赖于多个使用combine_latest(stream1, stream2, plot_function) 的 Observable 流的可视化。当一个 Observable 发生变化时，这很有效，比如当用户修改一个值时；使用新值更新可视化。但是，有时两个 Observable 会同时更新，例如当用户从单个文件加载两个流的数据时。从技术上讲，一个 Observable 将在另一个之前更新（以导入函数中的先到者为准）。因此，情节将被更新两次，但出于所有意图和目的，它只需要更新一次。

我拥有的一些可视化的计算成本很高，所以我想确保如果两个流同时更新，那么组合流只会发出一个值。我能想到几个选项：

1. 在组合流上使用 debounce() 并设置一个小的超时时间（如 50 毫秒）。这种方法对我来说似乎很脏。

2. 不要直接使用combine_latest。将两个流包装在一个新对象中，该对象也具有某种updating 标志。如果我将updating 标志设置为True，那么在我将updating 标志设置为False 之前不要发射任何东西。这种方法感觉是有状态的，它破坏了流的可组合性。

3. 告诉所有可视化在所有流都更新之前不要更新。同样，这打破了封装，因为可视化不应该关心上游发生了什么。它应该只是从组合流中接收新值并制作漂亮的图片。

4. 使可视化足够细化，以至于首先更新一个流只会带来很小的性能损失。这对于某些可视化来说是不可能的，例如基于点和网格大小计算网格的可视化。如果点或网格大小发生变化，则需要重新计算整个网格。

Rx 中是否有一些工具可以处理“同时”更新多个流？我觉得我要求的是魔法。

对于使用 Rx 制作 GUI 程序的任何人：除了通过流发送新值之外，我还应该为模型使用更好的架构吗？

如果这个问题不清楚，请在评论中告诉我，我会尝试做一个更具体的例子。

示例

这是一个示例 Python RxPY 程序：

import rx

stream1 = rx.subjects.BehaviorSubject(1)
stream2 = rx.subjects.BehaviorSubject(2)

rx.Observable\
  .combine_latest(stream1, stream2, lambda x, y: (x, y))\
  .subscribe(print)

stream1.on_next(3)
stream2.on_next(4)

打印出来：

(1, 2)
(3, 2)
(3, 4)

如何同时更新stream1 和stream2 的值，以使结果如下？

(1, 2)
(3, 4)

换句话说，我如何修改combine_latest，以便我可以在下游告诉它“嘿，等一下，我会在你发出下一个值之前更新其他流”？

Answer 1

我找到了一个可能的答案，但这不是最好的，我想要其他答案。

我发现了pausable 组合子。通过传入发出 True 或 False 的流，您可以控制是否暂停序列。这是我的示例的修改：

import rx

stream1 = rx.subjects.BehaviorSubject(1)
stream2 = rx.subjects.BehaviorSubject(2)

pauser = rx.subjects.BehaviorSubject(True)
rx.Observable\
         .combine_latest(stream1, stream2, lambda x, y: (x, y))\
         .pausable(pauser)\
         .subscribe(print)

# Begin updating simultaneously
pauser.on_next(False)
stream1.on_next(3)
stream2.on_next(4)

# Update done, resume combined stream
pauser.on_next(True)

# Prints:
# (1, 2)
# (3, 4)

要应用到我的 GUI，我可以在我的模型中创建一个名为 updating 的 BehaviorSubject，它会发出整个模型是否正在更新。例如，如果stream1 和stream2 同时更新，那么我可以将updating 设置为True。在任何制作成本高昂的可视化上，我都可以应用 updating 的值来暂停组合流。

Answer 2

这适用于 C# 的 Rx：

var throttled = source.Publish(hot => hot.Buffer(() => hot.Throttle(dueTime));

这里的dueTime 值是.NET 中的TimeSpan。它只是说明了您希望在它产生价值之前处于不活动状态的时间窗口是多少。这基本上会吞噬一段时间内“同时”产生的值。

在这种情况下，source 将是您的 .combine_latest(...) 可观察的。