为什么重复（n）不适用于反应式扩展中的创建

Question

from rx import Observable, Observer
from __future__ import print_function
import random

def create_observable(observer):
    while True:
        observer.on_next(random.randint(1,100))    

Observable.create(create_observable).take_while(lambda x: x>50).repeat(6).subscribe(print)

给予

74 78 94 59 79 76

序列，而我希望每个数字会重复 6 次

所以“重复”永远不会对使用 create 方法创建的 observables 起作用。

Answer 1

贴出的代码得到 6 次在 [51, 100] 范围内的随机整数序列。

试试

(Observable.create(create_observable)
    .take_while(lambda x: x > 50)
    .select_many(lambda x: Observable.just(x).repeat(6))
    .subscribe(print))

或者只是

(Observable.create(create_observable)
    .take_while(lambda x: x > 50)
    .select_many(lambda v: [v] * 6)
    .subscribe(print))

Answer 2

问题在于create 函数中的无限循环。每当您订阅此类 observable 时，它​​都会调用传递给 create 的函数，并且无法退出此循环。您的程序此时卡住了，没有进一步的代码执行。你没有提到，但我想你必须强行终止它。

生成无限序列时，您必须在每次迭代后检查订阅是否有效。最简单、最安全的方法是使用Observable.generate() 函数：

Observable.generate(random.randint(1, 100), 
                    lambda x: True,
                    lambda x: random.randint(1, 100),
                    lambda x: x) \
    .take_while(lambda x: x > 50) \
    .repeat(6) \
    .subscribe(print)