RxJS 基本介绍

RxJS 是 Reactive Extensions for JavaScript 的缩写，是一个基于可观测数据流 Stream 结合观察者模式和迭代器模式的一种异步编程的应用库。RxJS 是 Reactive Extensions 在 JavaScript 上的实现。

这里我们举一个例子，假如我们想要监听点击事件(click event)，但点击一次之后不再监听。

原生JavaScript

var handler = (e) => {
console.log(e);
document.body.removeEventListener('click', handler);
}

document.body.addEventListener('click', handler);

使用RxJs 大概的样子

import {fromEvent} from 'rxjs';
fromEvent(document.body, 'click').pipe(
  take(1) // 只取一次
).subscribe(console.log);

可以把 RxJS 想成处理 非同步行为 的 Lodash。(Think of RxJS as Lodash for events.)

这也被称为 Functional Reactive Programming，更确切的说是指 Functional Programming 及 Reactive Programming 两个编程思想的结合。

响应式编程(Reactive Programming)

响应式编程简单来说就是当变量或资源发生变动时，由变量或资源自动告诉我发生变动了

这句话看似简单，其实背后隐含两件事

• 当发生变动=> 非同步：不知道什么时候会发生变动，反正变动时要跟我说

• 由变动发生时自动告诉我 => 我不用写通知我的每一步代码,例如Vue2的双向绑定通过ES5 definedProperty 的getter/setter。每当变量发生变动时，就会执行getter/setter 从而收集有改动的变量，相关的变量及画面也会跟着变动，而开发者不需要关心这些变动如何发生的，这也被称为依赖收集。

Rx 基本上就是上述的两个观念的结合

函数式编程(Functional Programming)

简单说函数式编程核心思想就是做运算处理，并用函数来思考问题，例如像以下的算数运算式：

(5 + 6) - 1 * 3

我们可以写成

const add = (a, b) => a + b
const mul = (a, b) => a * b
const sub = (a, b) => a - b

sub(add(5, 6), mul(1, 3))

我们把每个运算包成一个个不同的function，并用这些function 组合出我们要的结果，这就是最简单的Functional Programming。

Functional Programming 基本条件

函数为一等公民(First Class)

一等公民就是指跟其他变量具有同等地位，也就是说函式能够被赋值给变量，函式也能够被当作参数传入另一个函数，也可当作一个函数的回传值

函数能够被赋值给变量

var hello = function() {}

函数能被当作参数传入

fetch('www.google.com')
.then(function(response) {}) // 匿名 function 传入 then()

函式能被当作回传值

var a = function(a) {
    return function(b) {
    return a + b;
  };  
}

Pure Function

Pure function 是指一个function 给予相同的参数，永远会返回相同的返回值，并且没有任何显著的副作用(Side Effect)

var arr = [1, 2, 3, 4, 5];

arr.slice(0, 3); // [1, 2, 3]

arr.slice(0, 3); // [1, 2, 3]

arr.slice(0, 3); // [1, 2, 3]

这里可以看到slice 不管执行几次，返回值都是相同的，并且除了返回一个值(value)之外并没有做任何事，所以 slice就是一个pure function。

var arr = [1, 2, 3, 4, 5];

arr.splice(0, 3); // [1, 2, 3]

arr.splice(0, 3); // [4, 5]

arr.slice(0, 3); // []

这里我们换成用 splice，因为 splice 每执行一次就会影响 arr 的值，导致每次结果都不同，这就很明显不是一个 pure function。

副作用(Side Effect)

副作用 是指一个function 做了跟本身运算返回值没有关系的事，比如说修改某个全域变量，或是修改传入参数的值，甚至是执行 console.log

Functional Programming 强调没有Side Effect，也就是function 要保持纯粹，只做运算并返回一个值，没有其他额外的行为。

引用透明

前面提到的pure function 不管外部环境如何，只要参数相同，函式执行的返回结果必定相同。这种不依赖任何外部状态，只依赖于传入的参数的特性也称为引用透明(Referential transparency)

Functional Programming 优势

可读性高
当我们透过一系列的函数处理数据的操作过程，代码能变得非常的简洁且可读性极高，例如下面的例子

[9, 4].concat([8, 7]) // 合并数组
.sort()  // 排序
.filter(x => x > 5) // 过滤

可维护性高
因为Pure function 等特性，执行结果不依赖外部状态，且不会对外部环境有任何操作，使Functional Programming 能更好的排错及编写单元测试。

什么是Observable ?

要理解可观察对象(Observable) 之前，需要先了解两个设计模式， 迭代器模式 跟观察者模式。

观察者模式

观察者模式其实很常遇到，在许多API 的设计上都用了观察者模式 ，最简单的例子就是DOM 的事件监听

function clickHandler(event) {
console.log('user click!');
}

document.body.addEventListener('click', clickHandler)

这就是观察者模式，我们可以对某件事注册监听，并在事件发生时，自动执行我们注册的监听者(listener)。

迭代器模式

遍历器（Iterator）是这样一种机制。它是一种接口，为各种不同的数据结构提供统一的访问机制。任何数据结构只要部署 Iterator 接口，就可以完成遍历操作（即依次处理该数据结构的所有成员）。

var arr = [1, 2, 3];

var iterator = arr[Symbol.iterator]();

iterator.next();
// { value: 1, done: false }
iterator.next();
// { value: 2, done: false }
iterator.next();
// { value: 3, done: false }
iterator.next();
// { value: undefined, done: true }

迭代器模式有两个优势
第一 它渐进式取得数据的特性可以拿来做延迟运算(Lazy evaluation)，让我们能用它来处理数据量比较多的情况。
第二 因为iterator 本身是遍历器接口，所以可以用所有数组的运算方法像map, filter

延迟运算(Lazy evaluation)

延迟运算是一种运算策略，简单来说我们延迟一个表达式的运算时机直到真正需要它的值在做运算。

function* getNumbers(words) {
    for (let word of words) {
        if (/^[0-9]+$/.test(word)) {
           yield parseInt(word, 10);
        }
    }
}


const iterator = getNumbers('123');

iterator.next();
// { value: 1, done: false }
iterator.next();
// { value: 2, done: false }
iterator.next();
// { value: 3, done: false }
iterator.next();
// { value: undefined, done: true }

这里我们没有立即运算，直到每次执行next()的时候才执行运算，这就是延迟运算(evaluation strategy)

在了解 Observer 跟 Iterator 后，发现其实 Observer 跟 Iterator 有个共通的特性，就是他们都是 渐进式(progressive) 的取得数据，差別只在于 Observer 是生产者(Producer)推送数据(push)，而 Iterator 是消费者(Consumer)拉取数据(pull) 而 Observable 就是这两个思想的结合,Observable 具备生产者推送数据的特性，同时能像数组，拥有处理数组的方法(map, filter...） 更简单的来说，Observable 就像是一个序列，一个可观测的数据流,里面的元素会随着时间推送。

如何创建可观察对象Observable

import { Observable } from 'rxjs';

const observable = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  setTimeout(() => {
    subscriber.next(4);
    subscriber.complete();
  }, 1000);
});

console.log('just before subscribe');

observable.subscribe({
  next(x) { console.log('got value ' + x); },
  error(err) { console.error('something wrong occurred: ' + err); },
  complete() { console.log('done'); }
});

console.log('just after subscribe');

just before subscribe
got value 1
got value 2
got value 3
just after subscribe
got value 4
done

观察者Observer

一个回调函数的集合，它知道如何去监听由Observable提供的值。它负责观察任务执行的状态并向流中发射信号。

const observer = {
  next: function(value) {
  console.log(value);
    },
  error: function(error) {
    console.log(error)
  },
  complete: function() {
    console.log('complete')
  }
}

• next：每当Observable 发送出新的值，next 方法就会被触发。

• complete：在Observable 没有其他的数据可以取得时，complete 方法就会被触发，在complete 被触发之后，next 方法就不会再起作用。

• error：每当Observable 内发生错误时，error 方法就会被触发。

Creation Operator

Observable 有很多创建实例的方法，下面列出 RxJS一些常用的

• of
• from
• fromEvent
• interval
• timer

from
从一个数组、类数组对象、Promise、迭代器对象或者类 Observable 对象创建一个 Observable.

import { from } from 'rxjs';

var arr = ['Jerry', 'Anna', 2016, 2017, '30 days']
var source = from(arr);

source.subscribe({
    next: function(value) {
        console.log(value)
    },
    complete: function() {
        console.log('complete!');
    },
    error: function(error) {
        console.log(error)
    }
});

// Jerry
// Anna
// 2016
// 2017
// 30 days
// complete!

of
与from的能力差不太多，只不过在使用的时候是传入一个一个参数来调用的，有点类似于js中的concat方法。同样也会返回一个Observable，它会依次将你传入的参数合并并将数据以同步的方式发出。

import { of } from 'rxjs';

of(10, 20, 30)
    .subscribe({
        next: value => console.log('next:', value),
        error: err => console.log('error:', err),
        complete: () => console.log('the end'),
    });

// Outputs
// next: 10
// next: 20
// next: 30
// the end

fromEvent
创建一个 Observable，该 Observable 发出来自给定事件对象的指定类型事件。可用于浏览器环境中的Dom事件或Node环境中的EventEmitter事件等。

import { fromEvent } from 'rxjs';

const clicks = fromEvent(document, 'click');
clicks.subscribe(x => console.log(x));

// Results in:
// MouseEvent object logged to console every time a click
// occurs on the document.

interval
使用该操作符创建的Observable可以在指定时间内发出连续的数字，和setInterval差不多。在我们需要获取一段连续的数字时，或者需要定时做一些操作时都可以使用该操作符实现我们的需求。

import { interval } from 'rxjs';
import { take } from 'rxjs/operators';

const numbers = interval(1000);

const takeFourNumbers = numbers.pipe(take(4));

takeFourNumbers.subscribe(x => console.log('Next: ', x));

// Logs:
// Next: 0
// Next: 1
// Next: 2
// Next: 3

Subscription (订阅)

Subscription就是表示Observable的执行，可以被清理。这个对象最常用的方法就是unsubscribe方法，它不需要任何参数，只是用来清理由Subscription占用的资源。同时，它还有add方法可以使我们取消多个订阅。

import { interval } from 'rxjs';

const observable1 = interval(400);
const observable2 = interval(300);

const subscription = observable1.subscribe(x => console.log('first: ' + x));
const childSubscription = observable2.subscribe(x => console.log('second: ' + x));

subscription.add(childSubscription);

setTimeout(() => {
// Unsubscribes BOTH subscription and childSubscription
subscription.unsubscribe();
}, 1000);

Marble diagrams(弹珠图)

我们在表达事物时，文字其实是最糟糕的手段。我们可以通过各种图示让我们更方便的理解observable 的各种operators

我们把描绘observable 的图示称为弹珠图(Marble diagrams)，RxJS 有非常多的弹珠图，规则大致上都是相同的。
我们用 - 來表达一小段时间，這些 - 串起就代表一个observable。

----------------

X (大写X)则代表有错误发生

---------------X

| 则代表observable 结束

----------------|

在这个时间序列中，我们可能会发送值(value)，这里我们用 interval 举例

var source = interval(1000);

source 的图形就会长像这样

-----0-----1-----2-----3--...

当observable 是同步送值的时候，例如

var source = of(1,2,3,4);

source 的图形就会长像这样

(1234)|

小括号代表着同步发生。

另外的Marble diagrams 也能够表达operator 的前后转换，例如

var source = interval(1000);
var newest = source.pipe(map(x => x + 1));

source: -----0-----1-----2-----3--...
map(x => x + 1)
newest: -----1-----2-----3-----4--...

最上面是原本的observable，中间是operator，下面则是新的observable。

以上就是Marble diagrams 如何表示operator 对observable 的操作，这能让我们更好的理解各个operator。

Operator概念

采用函数式编程风格的纯函数 (pure function)，使用像 map、filter、concat、flatMap 等这样的操作符来处理集合。也正因为他的纯函数定义，所以我们可以知道调用任意的操作符时都不会改变已存在的Observable实例，而是会在原有的基础上返回一个新的Observable。

尽管 RxJS 的根基是 Observable，但最有用的还是它的操作符。操作符是允许复杂的异步代码以声明式的方式进行轻松组合的基础代码单元。

map

Observable 的map 方法使用上跟数组的map 是一样的，我们传入一个callback function，这个callback function 会带入每次发送出来的元素，然后我们返回新的元素，如下

var source = interval(1000);
var newest = source.pipe(map(x => x + 1));

newest.subscribe(console.log);
// 2
// 3
// 4
// 5..

用Marble diagrams 表达就是

source: -----0-----1-----2-----3--...
map(x => x + 1)
newest: -----1-----2-----3-----4--...

mapTo

mapTo 可以把传进来的值改成一个固定的值，如下

var source = interval(1000);
var newest = source.pipe(mapTo(2));

newest.subscribe(console.log);
// 2
// 2
// 2
// 2..

source: -----0-----1-----2-----3--...
mapTo(2)
newest: -----2-----2-----2-----2--...

filter

var source = interval(1000);
var newest = source.pipe(filter(x => x % 2 === 0));

newest.subscribe(console.log);
// 0
// 2
// 4
// 6..

source: -----0-----1-----2-----3-----4-...
filter(x => x % 2 === 0)
newest: -----0-----------2-----------4-...

take

take 是一个很简单的operator，顾名思义就是取前几个元素后就结束

import { interval } from 'rxjs';
import { take } from 'rxjs/operators';

const intervalCount = interval(1000);
const takeFive = intervalCount.pipe(take(3));
takeFive.subscribe(x => console.log(x));

// Logs:
// 0
// 1
// 2

source : -----0-----1-----2-----3--..
take(3)
example: -----0-----1-----2|

skip

可以略过前几个送出元素的operator: skip，示例如下：

import { interval } from 'rxjs';
import { skip } from 'rxjs/operators';


const source = interval(1000);
const example = source.pipe(skip(3));
const subscribe = example.subscribe(val => console.log(val));
// 3
// 4
// 5...

source : ----0----1----2----3----4----5--....
skip(3)
example: -------------------3----4----5--...

first

first 会取observable 送出的第1 个元素之后就直接结束，行为跟take(1) 一致。

import { fromEvent } from 'rxjs';
import { first } from 'rxjs/operators';

const clicks = fromEvent(document, 'click');
const result = clicks.pipe(first());
result.subscribe(x => console.log(x));

source : -----0-----1-----2-----3--..
first()
example: -----0|

takeUntil

takeUntil 经常使用到，他可以在某件事情发生时，让一个observable 直接完成(complete)

import { fromEvent, interval } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

const source = interval(1000);
const clicks = fromEvent(document, 'click');
const result = source.pipe(takeUntil(clicks));
result.subscribe(x => console.log(x));

source : -----0-----1-----2------3--
click  : ----------------------c----
takeUntil(click)
example: -----0-----1-----2----|

当 click 一发送出元素的时候, observable 就会直接完成(complete)。

concat

concat可以把多个observable实例合并成一个，示例如下

var source = interval(1000).pipe(take(3))
var source2 = of(3)
var source3 = of(4,5,6)
var example = source.pipe(concat(source2, source3))
example.subscribe({
next: (value) => { console.log(value); },
error: (err) => { console.log('Error: ' + err); },
complete: () => { console.log('complete'); }
});
// 0
// 1
// 2
// 3
// 4
// 5
// 6
// complete

concat的行为永远都是先处理第一个observable，等到当前处理的结束后才会再处理下一个

concatAll

有时我们的Observable 送出的元素又是一个observable，就像是二维数组，数组里面的元素是数组，这时我们就可以用 concatAll铺平

var click = fromEvent(document.body, 'click');
var source = click.pipe(map(e => of(1,2,3)));

var example = source.pipe(concatAll());
example.subscribe({
next: (value) => { console.log(value); },
error: (err) => { console.log('Error: ' + err); },
complete: () => { console.log('complete'); }
});

click  : ------c------------c--------

map(e => Rx.Observable.of(1,2,3))

source : ------o------------o--------
\            \
(123)|       (123)|

concatAll()

example: ------(123)--------(123)------------

withLatestFrom

这个操作符有主从的关系，只有在主要的observable送出新的值时，才会执行callback，附属的observable只是在背后运行。让我们看一个例子

 const clicks = fromEvent(document, 'click');
 const timer = interval(1000);
 const result = clicks.pipe(withLatestFrom(timer));
 result.subscribe(x => console.log(x));

image.png

withLatestFrom 会在main 送出值的时候执行callback，但请注意如果main 传出值时some 之前没有送出过任何值callback 仍然不会执行

案例 拖拉demo

当我们在优酷看影片时往下滚动画面，视频会变成一个小视窗在右下角，这个视窗还能够拖拉移动位置。这个功能可以让使用者一边看留言同时又能看视频，且不影响其他的信息显示

需求分析

首先会有一个视频在最上方，网页滚动到低于视频高度后，视频改为fixed定位，往回滚会再变回原本的状态。当视频为fixed时，可以拖拉移动（drag），且移动范围不超过网页边缘

<div id="anchor">
    <div class="video" id="video">
        <div class="masker"></div>
        <video width="100%" controls>
            <source src="http://download.blender.org/peach/bigbuckbunny_movies/big_buck_bunny_480p_stereo.ogg"
                    type="video/ogg">
            Your browser does not support HTML5 video.
        </video>
    </div>
</div>

第一步，获取DOM

因为先做滚动切换class，所以这里用到的DOM只有#video，#anchor。

const video = document.getElementById('video');
const anchor = document.getElementById('anchor');

第二步，创建observable

这里做滚动效果，所以只需要监听滚动事件。

const scroll = fromEvent(document, 'scroll');

第三步，编写逻辑

判断是否滚过#anchor最底部

scroll.pipe(map(e => anchor.getBoundingClientRect().bottom < 0));

当我们可视范围区间滚过#anchor底部时，anchor.getBoundingClientRect（）.bottom就会变成负值，此时我们就改变#video的class。

scroll.pipe(
    map(e => anchor.getBoundingClientRect().bottom < 0)
).subscribe(bool => {
    if (bool) {
        video.classList.add('video-fixed');
    } else {
        video.classList.remove('video-fixed');
    }
})

const video = document.getElementById('video');
const anchor = document.getElementById('anchor');

const scroll = fromEvent(document, 'scroll');

scroll.pipe(
    map(e => anchor.getBoundingClientRect().bottom < 0)
).subscribe(bool => {
    if (bool) {
        video.classList.add('video-fixed');
    } else {
        video.classList.remove('video-fixed');
    }
})

接下来就可以接着做拖拽的行为

注册mousedown，mouseup，mousemove三个事件。

const mouseDown = fromEvent(video, 'mousedown')
const mouseUp = fromEvent(document, 'mouseup')
const mouseMove = fromEvent(document, 'mousemove')

首先点击#video，点击（mousedown）后要变成移动事件（mousemove），而移动事件会在鼠标放开（mouseup）时结束（takeUntil）

mouseDown.pipe(
    map(e => mouseMove.takeUntil(mouseUp)),
    concatAll()
)

因为把mouseDown observable发送出来的事件换成了mouseMove observable，所以变成了observable（mouseDown）送出observable（mouseMove）。因此最后用concatAll把后面送出的元素变成mouse move的事件。

因为我们的这段拖拽事件其实只能做用到video-fixed的时候，所以我们要加上filter

mouseDown.pipe(
    filter(e => video.classList.contains('video-fixed')),
    map(e => mouseMove.pipe(takeUntil(mouseUp))),
    concatAll(),
)

这里我们用filter如果当下#video没有video-dragable class的话，事件就不会送出。
再来我们就能把mousemove事件获取{ x，y }的坐标，并订阅来改变#video

mouseDown.pipe(
    filter(e => video.classList.contains('video-fixed')),
    map(e => mouseMove.pipe(takeUntil(mouseUp))),
    concatAll(),
    map(m => {
        return {
            x: m.clientX,
            y: m.clientY
        }
    }))
    .subscribe(pos => {
        video.style.top = pos.y + 'px';
        video.style.left = pos.x + 'px';
    })

解决闪烁问题
这个问题是因为我们的拖拉直接给元件鼠标的位置（clientX，clientY），而不是给鼠标相对元素
我们可以用withLatestFrom来把mousedown与mousemove两个Event的值同时传入callback。

mouseDown.pipe(
    filter(e => video.classList.contains('video-fixed')),
    map(e => mouseMove.pipe(takeUntil(mouseUp))),
    concatAll(),
    withLatestFrom(mouseDown, (move, down) => {
        return {
            x: move.clientX - down.offsetX,
            y: move.clientY - down.offsetY
        }
    })
).subscribe(pos => {
    video.style.top = pos.y + 'px';
    video.style.left = pos.x + 'px';
})

解决超出可视范围问题
拖拉会超出可视范围。这个问题其实只要给最大最小值就行了，因为元素是相对可视居间的绝对位置（fixed），也就是说

• top最小是0

• left最小是0

• top最大是可视高度扣掉元素本身高度

• left最大是可视宽度扣掉元素本身宽度

这里我们写一个function来处理这件事

const validValue = (value, max, min) => {
return Math.min(Math.max(value, min), max)
}

最终代码

mouseDown.pipe(
    filter(e => video.classList.contains('video-fixed')),
    map(e => mouseMove.pipe(takeUntil(mouseUp))),
    concatAll(),
    withLatestFrom(mouseDown, (move, down) => {
        return {
            x: validValue(move.clientX - down.offsetX, window.innerWidth - 320, 0),
            y: validValue(move.clientY - down.offsetY, window.innerHeight - 180, 0)
        }
    })
).subscribe(pos => {
    video.style.top = pos.y + 'px';
    video.style.left = pos.x + 'px';
})

我们简单地用了不到35行的代码，完成了一个还算复杂的功能。更重要的是我们还保持了整个程序的可读性，之后维护也更加的轻松。

深入 Observable

observable的operators跟数组一些方法的有很大的不同，主要差异有两点

• 延迟运算
• 渐进式取值
  延迟运算很好理解，所有Observable一定会等到订阅后才开始对元素做运算，如果没有订阅就不会有运算的行为

var source = from([1,2,3,4,5]);
var example = source.pipe(map(x => x + 1));

上面这段代码因为Observable还没有被订阅，所以不会真的对元素做运算，这跟数组的操作不一样，如下

var source = [1,2,3,4,5];
var example = source.map(x => x + 1);

上面这段代码执行完，example就已经取得所有元素的返回值了。

渐进式取值

数组的operators都必须完整的运算出每个元素的返回值并组成一个数组，再做下一个operator的运算，我们看下面这段代码

var source = [1,2,3]；
var example = source
.filter（x => x % 2 === 0）//这里会运算并返回一个完整的数组
.map（x => x + 1）//这里也会运算并返回一个完整的数组

上面这段代码，相信大家都很熟悉了，大家应该都有注意到source.filter（…）就会返回一整个新数组，再接下一个operator又会再返回一个新的数组

Observable operator的运算方式跟数组的是完全的不同，虽然Observable的operator也都会回传一个新的observable，但因为元素是渐进式取得的关系，所以每次的运算是一个元素运算到底，而不是运算完全部的元素再返回。

var source = Rx.Observable.from([1,2,3]);
var example = source.pipe(
  filter(x => x % 2 === 0)
  map(x => x + 1)
)

example.subscribe(console.log);

上面这段代码运行的方式是这样的
送出1到filter被过滤掉
送出2到filter在被送到map转成3，送到observer console.log印出
送出3到filter被过滤掉
每个元素送出后就是运算到底，在这个过程中不会等待其他的元素运算。这就是渐进式取值的特性

渐进式取值的观念在Observable中其实非常的重要，这个特性也使得Observable相较于Array的operator在做运算时来的高效很多，尤其是在处理大量数据的时候会非常明显

Subject

Subject 可以拿去订阅Observable(source) 代表他是一个Observer，同时Subject 又可以被Observer(observerA, observerB) 订阅，代表他是一个Observable。

总结成两句话
Subject 同时是Observable 又是Observer
Subject 会对内部的observers 清单进行多播(multicast)

使用 Subject 主要是为了多播（multicast）。Observable 默认是单播（unicast）的，而单播就意味着：对于每个订阅者，都只有一个独立的 Observable execution 与之对应。证明如下：

var source = interval(1000).pipe(take(3));

var observerA = {
next: value => console.log('A next: ' + value),
error: error => console.log('A error: ' + error),
complete: () => console.log('A complete!')
}

var observerB = {
next: value => console.log('B next: ' + value),
error: error => console.log('B error: ' + error),
complete: () => console.log('B complete!')
}

var subject = new Subject()

subject.subscribe(observerA)

source.subscribe(subject);

setTimeout(() => {
subject.subscribe(observerB);
}, 1000);

// "A next: 0"
// "A next: 1"
// "B next: 1"
// "A next: 2"
// "B next: 2"
// "A complete!"
// "B complete!"
// 代表这两次的订阅是完全分开来执行的，或者说是每次的订阅都建立了一个新的上下文。

由于 Observable 在设计上就是单播的，所以如果你希望使多个订阅者收到相同的数据，那么用 Observable 可能会非常麻烦。而 Subject 可以帮助我们解决这个问题。

Subject 也可比作事件发射器（EventEmitter），其中注册了多个事件监听器。 当我们订阅 Subject 时，它并不会启动一个新的 execution 来传送数据。而是在现有观察者列表中注册一个新的观察者，仅此而已。

var source = interval(1000).pipe(take(3));

var observerA = {
next: value => console.log('A next: ' + value),
error: error => console.log('A error: ' + error),
complete: () => console.log('A complete!')
}

var observerB = {
next: value => console.log('B next: ' + value),
error: error => console.log('B error: ' + error),
complete: () => console.log('B complete!')
}

var subject = new Subject()

subject.subscribe(observerA)

source.subscribe(subject);

setTimeout(() => {
subject.subscribe(observerB);
}, 1000);

// "A next: 0"
// "A next: 1"
// "B next: 1"
// "A next: 2"
// "B next: 2"
// "A complete!"
// "B complete!"

将我们的 subject 传递给 subscribe()，使其接收由 observable 传来的值（消费数据）。随后，subject 的所有订阅者都会立即收到这个值。

subject简易实现

var source = interval(1000).pipe(take(3))

var observerA = {
next: value => console.log('A next: ' + value),
error: error => console.log('A error: ' + error),
complete: () => console.log('A complete!')
}

var observerB = {
next: value => console.log('B next: ' + value),
error: error => console.log('B error: ' + error),
complete: () => console.log('B complete!')
}

var subject = {
observers: [],
subscribe: function(observer) {
this.observers.push(observer)
},
next: function(value) {
this.observers.forEach(o => o.next(value))
},
error: function(error){
this.observers.forEach(o => o.error(error))
},
complete: function() {
this.observers.forEach(o => o.complete())
}
}

subject.subscribe(observerA)

source.subscribe(subject);

setTimeout(() => {
subject. subscribe(observerB);
}, 1000);

// "A next: 0"
// "A next: 1"
// "B next: 1"
// "A next: 2"
// "B next: 2"
// "A complete!"
// "B complete!"