rxcpp

Question

我正在尝试了解如何使用 rxcpp，我的印象是当一个可观察对象发出一个值时，所有订阅的观察者都会通过调用他们的 on_next() 方法得到通知，传递它们是发射值。

以下示例不是这种情况：

auto eventloop = rxcpp::observe_on_event_loop();

printf("Start task\n");

auto values = rxcpp::observable<>::interval(std::chrono::seconds(2)).map(
        [](int i){
            printf("Observable sending: %d\n", i);
            return i;
        }
);

values.
    subscribe_on(eventloop).
    take(2).
    as_blocking().
    subscribe(
        [](int v){printf("#1 onNext: %d\n", v);},
        [](){printf("#1 onCompleted\n");});

values.
    subscribe_on(eventloop).
    take(2).
    as_blocking().
    subscribe(
        [](int v){printf("#2 onNext: %d\n", v);},
        [](){printf("#2 onCompleted\n");});

printf("Finish task\n");

我希望输出类似于：

Start task
Observable sending: 1
#1 onNext: 1
#2 onNext: 1
Observable sending: 2
#1 onNext: 2
#1 onCompleted
#2 onNext: 2
#2 onCompleted
Finish task

即on_next 在新值通过时调用所有订阅的观察者。

相反，输出实际上是：

Start task
Observable sending: 1
#1 onNext: 1
Observable sending: 2
#1 onNext: 2
#1 onCompleted
Observable sending: 1
#2 onNext: 1
Observable sending: 2
#2 onNext: 2
#2 onCompleted
Finish task

Answer 1

这是经典的热对冷行为。

热的 Observable 会如您所愿。 Interval 是一个冷可观察对象，因此每个订阅都会产生一组独立的值。

publish operator 将采用单个冷 observable 并将其作为热 observable 共享。

在这种情况下它会。

auto sharedvalues = values.publish().ref_count();

然后在订阅表达式中使用 sharedvalues 而不是 values。

搜索热 vs 冷 observables 会发现对这个主题的广泛讨论。

rxcpp - 当一个可观察对象发出一个值时，为什么不调用所有观察者的 on_next 函数

rxcpp - why don't all observers' on_next function get called when an observable emits a value

c++

reactive-programming

reactivex