摘要：最近有同学在使用时总是不能如愿拿到自己想要的数据，说到底还是没有能把思维从命令式的习惯中转换过来。响应式的编程风格中，数据应该在流内完成转换，合并，过滤，而不是取出来，一顿操作再丢回流里。
最近有同学在使用rxjs时总是不能如愿拿到自己想要的数据，说到底还是没有能把思维从命令式的习惯中转换过来。
Observable !== 异步 && Observable !== 同步
如题，请默念几遍！在（一）里面其实就提到过，Observable里的数据到底是同步的还是异步取决于你如何使用，这和promise是完全不同的，promise不论你如果使用，它始终是异步的。上代码：
dataSource: DataSource;
ngOnInit() {
this.getDataSource()
.subscribe(data => {
console.log("in subscribe: ",data);
this.dataSource = data;
})
console.log("after subscribe:", this.dataSource);
}
getDataSource(): Observable> {
return this.stockService.getStocksMat()
.pipe(
map(stocks => new StockTbDataSource(this.paginator, this.sort, stocks))
);
};
请问，以上2个console，哪个先输出，哪个后输出？正确答应该是，鬼知道！仔细分析代码，这个数据是从stockService上获取到的，那么数据是同步还是异步就取决于这个服务的 getStocksMat 方法。
import { of } from "rxjs";
export class StockService {
getStocksMat(): Observable> {
return of({name: "a", price: 100});
}
}
现在呢，毫无疑问，这个时候数据是同步的，因为服务的方法中使用of操作符创建了一个 Observable，of操作符的行为是会把传入它的参数依次推送到流上，最后发出完成通知。所以，以上两个console的输出顺序应该是 inner 先输出，after 后输出。
注意：也不要把操作符和同步异步划等号，同样也和你如何使用它有关系，这个系列里一直没有提到的一个东西叫 scheduler，也就是调度器，它可以调节流上的值如何发射
import { asyncScheduler } from "rxjs";
of({name: "a", price: 100}, asyncScheduler); // 此时流上的数据将被异步发出。
假如service上的代码变成：
export class StockService {
constructor(private http: HttpClient) { }
getStocksMat(): Observable> {
return this.http.get(someUrl).pipe(
map(res => res.data)
);
}
}
很显然，我们是想从服务器上取一段数据回来，那么这个过程肯定是异步的，所以那2个console的输出顺序就应该是 after 先输出，而inner 后输出。
尽量不要主动订阅流
这里指的是在angular里，因为angular给我们提供了取数据的 pipe - async。它可以帮助我们在模板中把数据取出来，当然就是订阅我们给它的 Observable。不主动订阅的原因如下：
当订阅流时，会产生 subscription，当然使用 async pipe 时也会有，但此时框架会帮我们管理它，当不需要再订阅时取消订阅，如模板销毁时。
如果我们手动订阅的是一个会发出结束通知的流时，rxjs的底层会帮我们在流上的数据发送完成时取消订阅，反之则不会。也就是说第一，你需要准确判断订阅的流是否会发出结束通知。第二，你可能需要在合适的时机手动取消订阅。
响应式的编程风格中，数据应该在流内完成转换，合并，过滤，而不是取出来，一顿操作再丢回流里。
如下
export class StockService {
constructor(private http: HttpClient) { }
getStocksMatArr(): Observable[]> {
return this.http.get(someUrl).pipe(
map(res => res.data)
);
}
// 只要价格大于某个值的股票
getStocksThatPriceLargeThan(price: number): Observable[]> {
return this.getStocksMatArr().pipe(
filter(stocks => stocks.filter(stock => stock.price > price))
)
}
// 和另外一些流上的数据组合，比如购买人数；
getStocksWithBuyCount(): Observable<{stocks: DataSource[]; count: number}> {
const count$ = of(2000);
return this.getStocksMatArr().pipe(
withLatestFrom(count$, (stocks, count) => ({stocks, count}))
);
}
// 当然还可以更复杂
showExample(): Observable {
return this.getStocksMatArr().pipe(
mergeMap(stocks => from(stocks)), // 先把stock逐个放到流上
filter(stock => stock.price < 50), // 过滤出来
take(10), // 拿前10支股票
withLatestFrom(obs), // 和另一条流的上数据组合
bufferCount(2), // 两两组合
reduce((acc, cur) => [...acc, cur], []) // 再合并起来
delay(2000), // 延迟2秒再发
... 等等， 一切取决于你的需求。
)
}
}
只要保持数据一直在流中，你就不必时时惦记着它到底是同步还异步，数据来了就消费，没来就一直等。刚开始时建议强迫自己不去订阅，这样才能很快的理解和适应响应式的风格。
文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。
转载请注明本文地址：https://www.ucloud.cn/yun/97258.html
