Python响应式(Reactive)编程条件执行

Question

我开始使用 RxPy 模块学习 Python 中的响应式(Reactive)编程。现在，我有一个案例，我想根据接收到的元素运行不同的功能。我用一种非常简单的方法实现了它

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

def do_one():
    print('exec one')

def do_two():
    print('exec two')

subject = Subject()
subject.pipe(
    ops.filter(lambda action: action == 'one')
).subscribe(lambda x: do_one())

subject.pipe(
    ops.filter(lambda action: action == 'two')
).subscribe(lambda x: do_two())


subject.on_next('one')
subject.on_next('two')

对我来说这似乎有点难看，但是，我无法在操作/可观察对象中找到任何案例、开关或其他方法来根据接收到的元素触发不同的执行。

有更好的方法吗？

Answer 1

这个解决方案可能不太好用，但您可以通过几种不同的方式使用它:

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

def do_one():
    print('exec one')

def do_two():
    print('exec two')

subject = Subject()
subject.subscribe(lambda x: globals()[x]())

subject.on_next('do_one')
subject.on_next('do_two')

产生

exec one
exec two

我也使用了类似的东西，在调用时类是未知的......像这样:

import Workers  # a module in my app with different workers

Wrkr = getattr(Workers, wrkr_type)  # in your case, x == wrkr_type
w = Wrkr(...)

所以当你需要将其中一些函数移动到不同的文件甚至模块中时，你仍然可以实现相同的解决方案......一般来说，当然，你不想使用 globals () ;)