angular2 学习笔记 ( rxjs 流 )

Posted 2020-08-09

RxJS 博大精深，看了好几篇文章都没有明白. 

范围牵扯到了函数响应式开发去了... 我对函数式一知半解, 响应式更是第一次听到... 

唉...不过日子还是得过...混着过先呗

我目前所理解的很浅, 大致上是这样的概念.

1.某些场景下它比promise好用, 它善于过滤掉不关心的东西. 

2.它是观察者模式 + 迭代器模式组成的 

3.跟时间，事件, 变量有密切关系

4.世界上有一种东西叫 "流" stream, 一个流能表示了一段时间里，一样东西发生的变化. 

  比如有一个值, 它在某段时间里从 "我" 变成 "你" 再变成 "他". 

  而我们可以对这个流进行观察，所以只要它发生变化，我们就会发现然后做任何事情。

5.站在游览器的角度, 服务器推送数据过来, 用户操作界面, timer 都是我们关心的流.

好，来看例子. 

我们通过 new Subject 来创建流. 也可以使用 new EventEmitter 或者 BehaviorSubject. 这些都继承了 Subject

EventEmitter 是 ng2 提供的

BehaviorSubject 可以填入初始值

import { Subject } from "rxjs/Subject";
private textEmitter: Subject<string> = new Subject(); 

要改变流中的值，我们使用 .next(value), 这个是迭代器的概念咯

keyup(value : string)
{
    this.textEmitter.next(value);
}

那么订阅是这样的 

ngOnInit() {
    this.text$ = this.textEmitter
        .debounceTime(500)
        .distinctUntilChanged()
        .switchMap(v => this.getDataAsync(v));

    this.text$.subscribe((value) => {
        console.log(value);
    });            
}

input keyup 性能优化, 我们通常会写一个 timeout + cleartimeout 的方式, 这个 debounceTime 就是干这个的 

流更新结束后 500ms 才会通知观察者 

distinctUntilChanged 是说只有当值和上一次通知时的值不一样的时候才通知观察者 

.map 和 .switchMap 都是用来对值进行处理的, 这个和 array.map 概念是一样的

而 .map 和 .switchMap 的区别是 .swichMap 处理那些返回 Observeable 的值 

getDataAsync(value : string): Observable<string>
{        
    let subject = new Subject();
    setTimeout(() => {
        console.log("after 2second");
        subject.next(value + "final");
    }, 2000);
    return subject;
}

如果我们使用 map 的话，它会直接返回 "subject" 这个对象, 而如果用 switchMap 它会返回这个 subject 对象的响应值.

<input type="text" #input (keyup)="keyup(input.value)" />
<p>{{ text$ | async }}</p>

ng2 提供了一个 async Pipe, 它会监听左边这个 text$ stream. 后面加一个 $ 符号通常用来表明这是一个 stream.

还有一个常用的功能是 combineLatest

就是可以同时监听多个流，只要其中一个有变动，那么所有的最新值都会发布出去, 可以用来实现依赖属性.

@Component({
    selector: "compute-property",
    template: ` 
        <input type="text" #input1 (keyup)="text1.next(input1.value)" />
        <input type="text" #input2 (keyup)="text2.next(input2.value)" />  
        {{ result$ | async }}                 
    `
})
export class ComputePropertyComponent implements OnInit {

    text1: BehaviorSubject<string> = new BehaviorSubject<string>("a");
    text2: BehaviorSubject<string> = new BehaviorSubject<string>("b");
    result$: Observable<string>;
    constructor() {}
    
    ngOnInit() {
        this.result$ = Observable.combineLatest(this.text1, this.text2).map(values => {          
            return values[0] + " " + values[1];
        }); 
    }     
}

还有 bufferCount, bufferTime 也是常用到

text: Subject<number> = new Subject<number>();
    
ngOnInit() {
    this.text.bufferCount(2)
        .subscribe(v => console.log(v)); //[v1,v2] 存够 count 了就发布

    this.text.bufferTime(2000)
        .subscribe(v => console.log(v)); //[v1,v2,...]把 2 秒内的所有 next value 放进来
}

Observable.of 可以简单的返回一个默认值 

Observable.of<string>("").subscribe(v => console.log(v));

rxjs 整个文档非常大，要按需加载.

通常做法是为项目开一个 rxjs-operators.ts 

import ‘rxjs/add/observable/throw‘; 
import ‘rxjs/add/observable/combineLatest‘; 
import ‘rxjs/add/observable/from‘; 
import ‘rxjs/add/observable/of‘; 
import ‘rxjs/add/operator/catch‘;
import ‘rxjs/add/operator/debounceTime‘;
import ‘rxjs/add/operator/distinctUntilChanged‘;
import ‘rxjs/add/operator/map‘;
import ‘rxjs/add/operator/switchMap‘;
import ‘rxjs/add/operator/toPromise‘;
import ‘rxjs/add/operator/startWith‘; 
import ‘rxjs/add/operator/bufferCount‘;
import ‘rxjs/add/operator/bufferTime‘;

放入常用的方法 

然后再 app.module.ts 里面导入它 

import ‘./rxjs-operators‘;