冷热观察：有“热”和“冷”运营商吗？ [英] Hot and Cold observables : are there 'hot' and 'cold' operators?

问题描述

我回顾了以下问题：

默认情况下，Observables很冷。订阅可观察量将导致上游订阅链发生。最后一个订阅导致执行一个函数，该函数将处理一个源并将其数据发送给它的观察者。

该观察者反过来发射给下一个观察者，结果在下游数据流中，直到接收器观察者。下面的简化插图显示了当两个订阅者订阅相同的observable时的订阅和数据流。

可以通过使用主题或通过多播运算符（及其派生词，参见下面的注释3）创建热观察值。

引擎盖下的多播运算符使用主题并返回可连接的observable。对运营商的所有订阅都将是对内部主题的订阅。当调用 connect 时，内部主题订阅上游observable，数据流向下游。
主体在内部操纵订阅观察者列表并将传入数据多播到所有订阅观察者。

下图总结了这种情况。

最后，了解由此引起的数据流更重要观察者模式和运算符的实现。

例如，如果 obs 很热，那么 hotOrCold = obs.op1 冷还是热？无论答案是什么：

• 如果没有 obs.op1 的订阅者，没有数据会流经 op1 。如果有热门 obs 的订阅者，则表示 obs.op1 可能会丢失数据


• 假设 op1 不是类似多播的运算符，订阅两次到 hotOrCold 将订阅两次到 op1 ，来自 obs 的每个值都会流过两次 op1 。

注意：

1. 此信息应对Rxjs v4有效。虽然版本5通过大量更改已经花费了b $ b，但大部分仍然是逐字适用。


2. 未表示取消订阅，错误和完成流程，因为
   它们不是在问题的范围内。调度程序也不考虑
   。除此之外，它们影响数据流的
   时间，但先验不是它的方向和内容。


3. 根据用于多播的主题类型，有
   不同的衍生多播运营商：


主题类型| `Publish`运算符| `Share`运算符
------------------ | --------------------------- | -----------------
Rx.Subject | Rx.Observable.publish |分享
Rx.BehaviorSubject | Rx.Observable.publishValue | shareValue
Rx.AsyncSubject | Rx.Observable.publishLast | N / A
Rx.ReplaySubject | Rx.Observable.replay | shareReplay


更新：另请参阅以下文章，这里 ，然后）。

进一步有关主题的详细信息可以在另一个SO问题中找到：什么是语义不同的RxJS科目？

I reviewed the following SO question: What are the Hot and Cold observables?

To summarize:

• a cold observable emits its values when it has an observer to consume them, i.e. the sequence of values received by observers is independent of time of subscription. All observers will consume the same sequence of values.
• a hot observable emits value independently of its subscriptions, i.e. the values received by observers are a function of the time of subscription.

Yet, I feel like hot vs. cold is still a source of confusion. So here are my questions:

• Are all rx observables cold by default (with the exception of subjects)?

  I often read that events are the typical metaphor for hot observables, but I also read that Rx.fromEvent(input, 'click') is a cold observable(?).

• Are there/what are the Rx operators which turn a cold observables into a hot observable (apart from publish, and share)?

  For instance, how does it work with Rx operator withLatestFrom? Let cold$ be a cold observable which has somewhere been subscribed to. Will sth$.withLatestFrom(cold$,...) be a hot observable?

  Or if I do sth1$.withLatestFrom(cold$,...), sth2$.withLatestFrom(cold$,...) and subscribe to sth1 and sth2, will I always see the same value for both sth?

• I thought Rx.fromEvent creates cold observables but that is not the case, as mentioned in one of the answers. However, I am still baffled by this behaviour : codepen.io/anon/pen/NqQMJR?editors=101. Different subscriptions get different values from the same observable. Wasn't the click event shared?

解决方案

I am coming back a few months later to my original question and wanted to share the gained knowledge in the meanwhile. I will use the following code as an explanation support (jsfiddle):

var ta_count = document.getElementById('ta_count');
var ta_result = document.getElementById('ta_result');
var threshold = 3;

function emits ( who, who_ ) {return function ( x ) {
  who.innerHTML = [who.innerHTML, who_ + " emits " + JSON.stringify(x)].join("\n");
};}

var messages$ = Rx.Observable.create(function (observer){
  var count= 0;
  setInterval(function(){
    observer.onNext(++count);
  }, 1000)
})
.do(emits(ta_count, 'count'))
.map(function(count){return count < threshold})
.do(emits(ta_result, 'result'))

messages$.subscribe(function(){});

As mentioned in one of the answers, defining an observable leads to a series of callback and parameter registration. The data flow has to be kicked in, and that is done via the subscribe function. A (simplified for illustration) detailed flow can be find thereafter.

Observables are cold by default. Subscribing to an observable will result in an upstream chain of subscriptions taking place. The last subscription leads to the execution of a function which will handle a source and emit its data to its observer.

That observer in its turn emits to the next observer, resulting in a downstream flow of data, down to the sink observer. The following simplified illustration shows subscription and data flows when two subscribers subscribe to the same observable.

Hot observables can be created either by using a subject, or through the multicast operator (and its derivatives, see Note 3 below).

The multicast operator under the hood makes use of a subject and returns a connectable observable. All subscriptions to the operator will be subscriptions to the inner subject. When connect is called, the inner subject subscribes to the upstream observable, and data flows downstream. Subjects manipulate internally a list of subscribed observers and multicast incoming data to all subscribed observers.

The following diagram summarizes the situation.

In the end, it matters more to understand the flow of data caused by the observer pattern and the implementation of the operators.

For instance, if obs is hot, is hotOrCold = obs.op1 cold or hot? Whatever the answer is :

• if there are no subscribers to obs.op1, no data will flow through op1. If there were subscribers to hot obs, that means obs.op1 will have possibly lost pieces of data
• supposing that op1 is not a multicast-like operator, subscribing twice to hotOrCold will subscribe twice to op1, and every value from obs will flow twice through op1.

Notes :

1. This information should be valid for Rxjs v4. While the version 5 has gone through considerable changes, most of it still applies verbatim.
2. Unsubscription, error and completion flows are not represented, as they are not in the scope of the question. Schedulers are also not taken into account. Among other things, they influence the timing of the data flow, but a priori not its direction and content.
3. According to the type of subject used for multicasting, there are different derived multicasting operators:

Subject type | `Publish` Operator | `Share` operator ------------------ | --------------------------- | ----------------- Rx.Subject | Rx.Observable.publish | share Rx.BehaviorSubject | Rx.Observable.publishValue | shareValue Rx.AsyncSubject | Rx.Observable.publishLast | N/A Rx.ReplaySubject | Rx.Observable.replay | shareReplay

Update : See also the following articles, here, and there) on that subject by Ben Lesh.

Further details on subjects can be found in this other SO question : What are the semantics of different RxJS subjects?

这篇关于冷热观察：有“热”和“冷”运营商吗？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！