python - rxpy 有效地组合可观察量

Question

简介:
你好。我正在探索 python rxpy用于我的用例的库 - 我正在使用响应式(Reactive)编程概念构建执行管道。通过这种方式，我希望我不必操纵太多状态。虽然我的解决方案似乎很实用，但是我在尝试从其他 Observable 组合一个新的 Observable 时遇到了麻烦。
问题是我编写观察值的方式导致一些昂贵的计算重复两次。为了性能，我真的想防止触发昂贵的计算。
我对响应式(Reactive)编程很陌生。试图挠头并查看互联网资源和引用文档 - 对我来说似乎有点太简洁了。请指教。
以下是说明我在做什么的玩具示例:

import rx
from rx import operators as op
from rx.subject import Subject

root = Subject()

foo = root.pipe(
        op.map( lambda x : x + 1 ),
        op.do_action(lambda r: print("foo(x) = %s (expensive)" % str(r)))
    )

bar_foo = foo.pipe(
        op.map( lambda x : x * 2 ),
        op.do_action(lambda r: print("bar(foo(x)) = %s" % str(r)))
    )

bar_foo.pipe(
        op.zip(foo),
        op.map(lambda i: i[0]+i[1]),
        op.do_action(lambda r: print("foo(x) + bar(foo(x)) = %s" % str(r)))
    ).subscribe()


print("-------------")
root.on_next(10)
print("-------------")

输出:

-------------
foo(x) = 11 (expensive)
bar(foo(x)) = 22
foo(x) = 11 (expensive)
foo(x) + bar(foo(x)) = 33
-------------

你可以想到 foo()和 bar()是昂贵和复杂的操作。我首先构建了一个可观察的 foo .然后组合一个新的 observable bar_foo包含 foo .稍后将两者压缩在一起以计算最终结果 foo(x)+bar(foo(x)) .
问题:

我能做些什么来防止 foo()从单个输入被触发多次？
我有非常充分的理由保留 foo()和 bar()分离。另外我也不想明确记住 foo() .

任何有在生产中使用 rxpy 经验的人都可以分享他们的经验。与等效的手工编写(但无法维护)的代码相比，使用 rxpy 会导致更好的性能或速度下降吗？

Answer 1

添加 op.share()紧接着 foo 中的昂贵计算管道在这里可能很有用。于是改了foo管道到:

foo = root.pipe(
        op.map( lambda x : x + 1 ),
        op.do_action(lambda r: print("foo(x) = %s (expensive)" % str(r))),
        op.share() # added to pipeline
    )

将导致:

-------------
foo(x) = 11 (expensive)
bar(foo(x)) = 22
foo(x) + bar(foo(x)) = 33
-------------

我相信 .share()使昂贵操作的发出事件在下游订阅者之间共享，以便单个昂贵计算的结果可以多次使用。
关于你的第二个问题；我也是 RxPy 的新手，因此对更有经验的用户的回答很感兴趣。到现在为止，我注意到作为初学者，您可以轻松创建(坏)管道，其中消息和计算在后台重复。 .share()似乎在一定程度上减少了这种情况，但不确定后台发生了什么。