如果说 Observable 与 observer 是攻受结合体的话，那么 Subject 就是一个人即攻亦受。正因为如此，我们在写一个Service用于数据传递时，总是使用 new Subject。
@Injectable() export class MessageService { private subject = new Subject<any>(); send(message: any) { this.subject.next(message); } get(): Observable<any> { return this.subject.asObservable(); } }
当F组件需要向M组件传递数据时，我们可以在F组件中使用 send()。
constructor(public srv: MessageService) { } ngOnInit() { this.srv.send('w s k f m?') }
而M组件只需要订阅内容就行：
constructor(private srv: MessageService) {} message: any; ngOnInit() { this.srv.get().subscribe((result) => { this.message = result; }) }
EventEmitter
其实EventEmitter跟RxJS没有直接关系，因为他是Angular的产物，而非RxJS的东西。或者我们压根没必要去谈，因为EventEmitter就是Subject。
EventEmitter的作用是使指令或组件能自定义事件。
@Output() changed = new EventEmitter<string>();
click() {
this.changed.emit('hi~');
}
@Component({ template: `<comp (changed)="subscribe($event)"></comp>` }) export class HomeComponent { subscribe(message: string) { // 接收：hi~ } }
上面示例其实和上一个示例中 MessageService 如出一辙，只不过是将 next() 换成 emit() 仅此而已。
结论
RxJS最难我想就是各种operator的应用了，这需要一些经验的积累。
RxJS很火很大原因我认还是提供了丰富的API，以下是摘抄：
创建数据流：
单值：of, empty, never
多值：from
定时：interval, timer
从事件创建：fromEvent
从Promise创建：fromPromise
自定义创建：create
转换操作：
改变数据形态：map, mapTo, pluck
过滤一些值：filter, skip, first, last, take
时间轴上的操作：delay, timeout, throttle, debounce, audit, bufferTime
累加：reduce, scan
异常处理：throw, catch, retry, finally
条件执行：takeUntil, delayWhen, retryWhen, subscribeOn, ObserveOn
转接：switch
组合数据流：
concat，保持原来的序列顺序连接两个数据流
merge，合并序列
race，预设条件为其中一个数据流完成
forkJoin，预设条件为所有数据流都完成
zip，取各来源数据流最后一个值合并为对象