python - 为什么 RxPY Observable 充当无限可迭代对象？

Question

我遇到了一个错误，它意外地将 Observable 用作可迭代对象。对于大多数对象，这通常很容易检测到:

>>> tuple(object())

Traceback (most recent call last):
  File "C:\Program Files (x86)\Python27\lib\site-packages\IPython\core\interactiveshell.py", line 3035, in run_code
    exec(code_obj, self.user_global_ns, self.user_ns)
  File "<ipython-input-4-40e3dfc60da8>", line 1, in <module>
    tuple(object())
TypeError: 'object' object is not iterable

但是，对于 Rx observable，它会默默地使 Python 崩溃:

<小时/>

MWE:

from rx import Observable

observable = Observable.from_list([1,2,3])
tuple(observable)  # Python will die silently here

没有回溯，也没有迹象表明存在任何问题。这使得本来就很难调试的并发响应式(Reactive)代码变得更加难以调试——我花了 2 个小时才最终找到这个代码。

<小时/>

经过仔细检查，迭代 Observable 似乎会创建新的 observable，尽管我不知道该 observable 没有 __iter__ 方法。

>>> for i, x in enumerate(observable):
>>>     print x
>>>     if i > 100:  # To prevent Python from crashing
>>>         break

<rx.anonymousobservable.AnonymousObservable object at 0x03111710>
<rx.anonymousobservable.AnonymousObservable object at 0x03111850>
<rx.anonymousobservable.AnonymousObservable object at 0x03111990>
<rx.anonymousobservable.AnonymousObservable object at 0x03111AD0>
<rx.anonymousobservable.AnonymousObservable object at 0x03111C10>
<rx.anonymousobservable.AnonymousObservable object at 0x03111D50>
<rx.anonymousobservable.AnonymousObservable object at 0x03111E90>
etc...

这是一个错误还是一个功能？ Observable 是否旨在可迭代？

Answer 1

我遇到了同样的问题，事实上，我因为不小心将可观察到的列表变成了列表而导致了我的计算机崩溃，这有点有趣。

我不是专家，一周前我就陷入了困境，但这似乎是一个功能。默认情况下，可观察量是非阻塞的，因此如果没有任何内容，它只会返回一个空的可观察量。这就像使用 O_NONBLOCK 标志打开的文件:读取调用立即返回一个空字符串。

如果您想将其用作阻塞迭代器，请使用to_blocking()。然后您可以执行以下操作:

from rx import Observable
o = Observable.from_([1, 2, 3, 4])
i = iter(o.to_blocking())
list(i)

它会完美地工作。

编辑:我刚刚发现为什么会发生这种情况。 Python 的迭代器协议(protocol)与 Observable 上的方法之一(next() 方法)之间存在冲突。文档说:

next() — returns an iterable that blocks until the Observable emits another item, then returns that item.

在 python 领域，next() 预计会返回可迭代对象上的下一个项目。因此，您可以获得无限量的可观察量，每个可观察量都 promise 在未来的某个日期返回一个项目。