mirror of
https://github.com/LCTT/TranslateProject.git
synced 2025-02-03 23:40:14 +08:00
PRF:20170709 The Extensive Guide to Creating Streams in RxJS.md
@BriFuture 翻译的很好。有部分地方我将英文翻译为中文了,对于这种前沿领域变化很快的部分,翻译跟不上技术变化。
This commit is contained in:
parent
a2e6402384
commit
119ea765c2
@ -1,22 +1,19 @@
|
||||
在 RxJS 中创建流的延伸教程
|
||||
============================================================
|
||||
全面教程:在 RxJS 中创建流
|
||||
================================
|
||||
|
||||
![](https://cdn-images-1.medium.com/max/900/1*hj8mGnl5tM_lAlx5_vqS-Q.jpeg)
|
||||
|
||||
对大多数开发者来说,RxJS 是以库的形式与之接触,就像 Angular。一些函数会返回流,要使用它们就得把注意力放在操作符上。
|
||||
对大多数开发者来说,与 RxJS 的初次接触是通过库的形式,就像 Angular。一些函数会返回<ruby>流<rt>stream</rt></ruby>,要使用它们就得把注意力放在操作符上。
|
||||
|
||||
有些时候,混用响应式和非响应式代码似乎很有用。然后大家就开始热衷流的创造。不论是在编写异步代码或者是数据处理时,流都是一个不错的方案。
|
||||
|
||||
RxJS 提供很多方式来创建流。不管你遇到的是什么情况,都会有一个完美的创建流的方式。你可能根本用不上它们,但了解它们可以节省你的时间,让你少码一些代码。
|
||||
|
||||
我把所有可能的方法,按它们的主要目的,分放在四个目录中:
|
||||
我把所有可能的方法,按它们的主要目的,放在四个分类当中:
|
||||
|
||||
* 流式化现有数据
|
||||
|
||||
* 生成数据
|
||||
|
||||
* 使用现有 APIs 进行交互
|
||||
|
||||
* 使用现有 API 进行交互
|
||||
* 选择现有的流,并结合起来
|
||||
|
||||
注意:示例用的是 RxJS 6,可能会以前的版本有所不同。已知的区别是你导入函数的方式不同了。
|
||||
@ -25,9 +22,7 @@ RxJS 6
|
||||
|
||||
```
|
||||
import {of, from} from 'rxjs';
|
||||
```
|
||||
|
||||
```
|
||||
of(...);
|
||||
from(...);
|
||||
```
|
||||
@ -38,36 +33,24 @@ RxJS < 6
|
||||
import { Observable } from 'rxjs/Observable';
|
||||
import 'rxjs/add/observable/of';
|
||||
import 'rxjs/add/observable/from';
|
||||
```
|
||||
|
||||
```
|
||||
Observable.of(...);
|
||||
Observable.from(...);
|
||||
```
|
||||
|
||||
```
|
||||
//or
|
||||
```
|
||||
//或
|
||||
|
||||
```
|
||||
import { of } from 'rxjs/observable/of';
|
||||
import { from } from 'rxjs/observable/from';
|
||||
```
|
||||
|
||||
```
|
||||
of(...);
|
||||
from(...);
|
||||
```
|
||||
|
||||
流的图示中的标记:
|
||||
|
||||
* | 表示流结束了
|
||||
|
||||
* X 表示流出现错误并被终结
|
||||
|
||||
* … 表示流的走向不定
|
||||
|
||||
* * *
|
||||
* `|` 表示流结束了
|
||||
* `X` 表示流出现错误并被终结
|
||||
* `...` 表示流的走向不定
|
||||
|
||||
### 流式化已有数据
|
||||
|
||||
@ -75,7 +58,7 @@ from(...);
|
||||
|
||||
#### of
|
||||
|
||||
如果只有一个或者一些不同的元素,使用 _of_ :
|
||||
如果只有一个或者一些不同的元素,使用 `of`:
|
||||
|
||||
```
|
||||
of(1,2,3)
|
||||
@ -89,13 +72,11 @@ of(1,2,3)
|
||||
|
||||
#### from
|
||||
|
||||
如果有一个数组或者 _可迭代的_ 对象,而且你想要其中的所有元素发送到流中,使用 _from_。你也可以用它来把一个 promise 对象变成可观测的。
|
||||
如果有一个数组或者 _可迭代的对象_ ,而且你想要其中的所有元素发送到流中,使用 `from`。你也可以用它来把一个 promise 对象变成可观测的。
|
||||
|
||||
```
|
||||
const foo = [1,2,3];
|
||||
```
|
||||
|
||||
```
|
||||
from(foo)
|
||||
.subscribe();
|
||||
```
|
||||
@ -111,9 +92,7 @@ from(foo)
|
||||
|
||||
```
|
||||
const foo = { a: 1, b: 2};
|
||||
```
|
||||
|
||||
```
|
||||
pairs(foo)
|
||||
.subscribe();
|
||||
```
|
||||
@ -125,19 +104,16 @@ pairs(foo)
|
||||
|
||||
#### 那么其他的数据结构呢?
|
||||
|
||||
也许你的数据存储在自定义的结构中,而它又没有实现 _Iterable_ 接口,又或者说你的结构是递归的,树状的。也许下面某种选择适合这些情况:
|
||||
也许你的数据存储在自定义的结构中,而它又没有实现 _可迭代的对象_ 接口,又或者说你的结构是递归的、树状的。也许下面某种选择适合这些情况:
|
||||
|
||||
* 先将数据提取到数组里
|
||||
1. 先将数据提取到数组里
|
||||
2. 使用下一节将会讲到的 `generate` 函数,遍历所有数据
|
||||
3. 创建一个自定义流(见下一节)
|
||||
4. 创建一个迭代器
|
||||
|
||||
* 使用下一节将会讲到的 _generate_ 函数,遍历所有数据
|
||||
稍后会讲到选项 2 和 3 ,因此这里的重点是创建一个迭代器。我们可以对一个 _可迭代的对象_ 调用 `from` 创建一个流。 _可迭代的对象_ 是一个对象,可以产生一个迭代器(如果你对细节感兴趣,参考 [这篇 mdn 文章][6])。
|
||||
|
||||
* 创建一个自定义流(见下一节)
|
||||
|
||||
* 创建一个迭代器
|
||||
|
||||
稍后会讲到选项 2 和 3 ,因此这里的重点是创建一个迭代器。我们可以对一个 _iterable_ 对象调用 _from_ 创建一个流。 _iterable_ 是一个对象,可以产生一个迭代器(如果你对细节感兴趣,参考 [这篇 mdn 文章][6])。
|
||||
|
||||
创建一个迭代器的简单方式是 [generator function][7]。当你调用一个生成函数(generator function)时,它返回一个对象,该对象同时遵循 _iterable_ 接口和 _iterator_ 接口。
|
||||
创建一个迭代器的简单方式是 <ruby>[生成函数][7]<rt>generator function</rt></ruby>。当你调用一个生成函数时,它返回一个对象,该对象同时遵循 _可迭代的对象_ 接口和 _迭代器_ 接口。
|
||||
|
||||
```
|
||||
// 自定义的数据结构
|
||||
@ -147,23 +123,17 @@ class List {
|
||||
get size() ...
|
||||
...
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
function* listIterator(list) {
|
||||
for (let i = 0; i<list.size; i++) {
|
||||
yield list.get(i);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
const myList = new List();
|
||||
myList.add(1);
|
||||
myList.add(3);
|
||||
```
|
||||
|
||||
```
|
||||
from(listIterator(myList))
|
||||
.subscribe(console.log);
|
||||
```
|
||||
@ -173,15 +143,13 @@ from(listIterator(myList))
|
||||
// 1 3 |
|
||||
```
|
||||
|
||||
调用 `listIterator` 函数时,返回值是一个 _iterable_ / _iterator_。函数里面的代码在调用 _subscribe_ 前不会执行。
|
||||
|
||||
* * *
|
||||
调用 `listIterator` 函数时,返回值是一个 _可迭代的对象_ / _迭代器_ 。函数里面的代码在调用 `subscribe` 前不会执行。
|
||||
|
||||
### 生成数据
|
||||
|
||||
你知道要发送哪些数据,但想(或者不得不)动态生成它。所有函数的最后一个参数都可以用来接收一个调度器。他们产生静态的流。
|
||||
你知道要发送哪些数据,但想(或者必须)动态生成它。所有函数的最后一个参数都可以用来接收一个调度器。他们产生静态的流。
|
||||
|
||||
#### range
|
||||
#### 范围(`range`)
|
||||
|
||||
从初始值开始,发送一系列数字,直到完成了指定次数的迭代。
|
||||
|
||||
@ -195,9 +163,9 @@ range(10, 2) // 从 10 开始,发送两个值
|
||||
// 10 11 |
|
||||
```
|
||||
|
||||
#### 间隔/定时器
|
||||
#### 间隔(`interval`) / 定时器(`timer`)
|
||||
|
||||
有点像 _range_,但定时器是周期性的发送累加的数字(就是说,不是立即发送)。两者的区别在于在于 _timer_ 允许你为第一个元素设定一个延迟。也可以只产生一个值,只要不指定周期。
|
||||
有点像范围,但定时器是周期性的发送累加的数字(就是说,不是立即发送)。两者的区别在于在于定时器允许你为第一个元素设定一个延迟。也可以只产生一个值,只要不指定周期。
|
||||
|
||||
```
|
||||
interval(1000) // 每 1000ms = 1 秒 发送数据
|
||||
@ -211,9 +179,7 @@ interval(1000) // 每 1000ms = 1 秒 发送数据
|
||||
|
||||
```
|
||||
delay(5000, 1000) // 和上面相同,在开始前先等待 5000ms
|
||||
```
|
||||
|
||||
```
|
||||
delay(5000)
|
||||
.subscribe(i => console.log("foo");
|
||||
// 5 秒后打印 foo
|
||||
@ -229,7 +195,7 @@ interval(10000).pipe(
|
||||
|
||||
这段代码每 10 秒获取一次数据,更新屏幕。
|
||||
|
||||
#### generate
|
||||
#### 生成(`generate `)
|
||||
|
||||
这是个更加复杂的函数,允许你发送一系列任意类型的对象。它有一些重载,这里你看到的是最有意思的部分:
|
||||
|
||||
@ -246,15 +212,13 @@ generate(
|
||||
// 1 2 4 8 |
|
||||
```
|
||||
|
||||
你也可以用它来迭代值,如果一个结构没有实现 _Iterable_ 接口。我们用前面的 list 例子来进行演示:
|
||||
你也可以用它来迭代值,如果一个结构没有实现 _可迭代的对象_ 接口。我们用前面的列表例子来进行演示:
|
||||
|
||||
```
|
||||
const myList = new List();
|
||||
myList.add(1);
|
||||
myList.add(3);
|
||||
```
|
||||
|
||||
```
|
||||
generate(
|
||||
0, // 从这个值开始
|
||||
i => i < list.size, // 条件:发送数据,直到遍历完整个列表
|
||||
@ -268,15 +232,13 @@ generate(
|
||||
// 1 3 |
|
||||
```
|
||||
|
||||
如你所见,我添加了另一个参数:选择器(selector)。它和 _map_ 操作符作用类似,将生成的值转换为更有用的东西。
|
||||
|
||||
* * *
|
||||
如你所见,我添加了另一个参数:选择器。它和 `map` 操作符作用类似,将生成的值转换为更有用的东西。
|
||||
|
||||
### 空的流
|
||||
|
||||
有时候你要传递或返回一个不用发送任何数据的流。有三个函数分别用于不同的情况。你可以给这三个函数传递调度器。_empty_ 和 _throwError_ 接收一个调度器参数。
|
||||
有时候你要传递或返回一个不用发送任何数据的流。有三个函数分别用于不同的情况。你可以给这三个函数传递调度器。`empty` 和 `throwError` 接收一个调度器参数。
|
||||
|
||||
#### empty
|
||||
#### `empty`
|
||||
|
||||
创建一个空的流,一个值也不发送。
|
||||
|
||||
@ -290,7 +252,7 @@ empty()
|
||||
// |
|
||||
```
|
||||
|
||||
#### never
|
||||
#### `never`
|
||||
|
||||
创建一个永远不会结束的流,仍然不发送值。
|
||||
|
||||
@ -304,7 +266,7 @@ never()
|
||||
// ...
|
||||
```
|
||||
|
||||
#### throwError
|
||||
#### `throwError`
|
||||
|
||||
创建一个流,流出现错误,不发送数据。
|
||||
|
||||
@ -318,15 +280,13 @@ throwError('error')
|
||||
// X
|
||||
```
|
||||
|
||||
* * *
|
||||
|
||||
### 挂钩已有的 API
|
||||
|
||||
不是所有的库和所有你之前写的代码使用或者支持流。幸运的是 RxJS 提供函数用来桥接非响应式和响应式代码。这一节仅仅讨论 RxJS 为桥接代码提供的模版。
|
||||
|
||||
你可能还对这篇出自 [Ben Lesh][9] 的 [延伸阅读][8] 感兴趣,这篇文章讲了几乎所有能与 promises 交互操作的方式。
|
||||
你可能还对这篇出自 [Ben Lesh][9] 的 [全面的文章][8] 感兴趣,这篇文章讲了几乎所有能与 promises 交互操作的方式。
|
||||
|
||||
#### from
|
||||
#### `from`
|
||||
|
||||
我们已经用过它,把它列在这里是因为,它可以封装一个含有 observable 对象的 promise 对象。
|
||||
|
||||
@ -346,9 +306,7 @@ fromEvent 为 DOM 元素添加一个事件监听器,我确定你知道这个
|
||||
|
||||
```
|
||||
const element = $('#fooButton'); // 从 DOM 元素中创建一个 jQuery 对象
|
||||
```
|
||||
|
||||
```
|
||||
from(element, 'click')
|
||||
.subscribe();
|
||||
```
|
||||
@ -367,31 +325,25 @@ from(document, 'click')
|
||||
.subscribe();
|
||||
```
|
||||
|
||||
这告诉 RxJS 我们想要监听 document 中的点击事件。在提交过程中,RxJS 发现 document 是一个 _EventTarget_ 类型,因此它可以调用它的 _addEventListener_ 方法。如果我们传入的是一个 jQuery 对象而非 document,那么 RxJs 知道它得调用 _on_ 方法。
|
||||
这告诉 RxJS 我们想要监听 document 中的点击事件。在提交过程中,RxJS 发现 document 是一个 _EventTarget_ 类型,因此它可以调用它的 `addEventListener` 方法。如果我们传入的是一个 jQuery 对象而非 document,那么 RxJs 知道它得调用 _on_ 方法。
|
||||
|
||||
这个例子用的是 _fromEventPattern_,和 _fromEvent_ 的工作基本上一样:
|
||||
这个例子用的是 _fromEventPattern_ ,和 _fromEvent_ 的工作基本上一样:
|
||||
|
||||
```
|
||||
function addClickHandler(handler) {
|
||||
document.addEventListener('click', handler);
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
function removeClickHandler(handler) {
|
||||
document.removeEventListener('click', handler);
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
fromEventPattern(
|
||||
addClickHandler,
|
||||
removeClickHandler,
|
||||
)
|
||||
.subscribe(console.log);
|
||||
```
|
||||
|
||||
```
|
||||
// 等效于
|
||||
fromEvent(document, 'click')
|
||||
```
|
||||
@ -402,49 +354,37 @@ RxJS 自动创建实际的监听器( _handler_ )你的工作是添加或者
|
||||
|
||||
```
|
||||
const listeners = [];
|
||||
```
|
||||
|
||||
```
|
||||
class Foo {
|
||||
registerListener(listener) {
|
||||
listeners.push(listener);
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
emit(value) {
|
||||
listeners.forEach(listener => listener(value));
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
const foo = new Foo();
|
||||
```
|
||||
|
||||
```
|
||||
fromEventPattern(listener => foo.registerListener(listener))
|
||||
.subscribe();
|
||||
```
|
||||
|
||||
```
|
||||
foo.emit(1);
|
||||
```
|
||||
|
||||
```
|
||||
// Produces
|
||||
// 结果
|
||||
// 1 ...
|
||||
```
|
||||
|
||||
当我们调用 foo.emit(1) 时,RxJS 中的监听器将被调用,然后它就能把值发送到流中。
|
||||
当我们调用 `foo.emit(1)` 时,RxJS 中的监听器将被调用,然后它就能把值发送到流中。
|
||||
|
||||
你也可以用它来监听多个事件类型,或者结合所有可以通过回调进行通讯的 API,例如,WebWorker API:
|
||||
|
||||
```
|
||||
const myWorker = new Worker('worker.js');
|
||||
```
|
||||
|
||||
```
|
||||
fromEventPattern(
|
||||
handler => { myWorker.onmessage = handler },
|
||||
handler => { myWorker.onmessage = undefined }
|
||||
@ -465,20 +405,14 @@ fromEventPattern(
|
||||
function foo(value, callback) {
|
||||
callback(value);
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
// 没有流
|
||||
foo(1, console.log); //prints 1 in the console
|
||||
```
|
||||
|
||||
```
|
||||
// 有流
|
||||
const reactiveFoo = bindCallback(foo);
|
||||
// 当我们调用 reactiveFoo 时,它返回一个 observable 对象
|
||||
```
|
||||
|
||||
```
|
||||
reactiveFoo(1)
|
||||
.subscribe(console.log); // 在控制台打印 1
|
||||
```
|
||||
@ -494,51 +428,39 @@ reactiveFoo(1)
|
||||
|
||||
```
|
||||
import { webSocket } from 'rxjs/webSocket';
|
||||
```
|
||||
|
||||
```
|
||||
let socket$ = webSocket('ws://localhost:8081');
|
||||
```
|
||||
|
||||
```
|
||||
// 接收消息
|
||||
socket$.subscribe(
|
||||
(msg) => console.log('message received: ' + msg),
|
||||
(err) => console.log(err),
|
||||
() => console.log('complete') * );
|
||||
```
|
||||
|
||||
```
|
||||
// 发送消息
|
||||
socket$.next(JSON.stringify({ op: 'hello' }));
|
||||
```
|
||||
|
||||
把 websocket 功能添加到你的应用中真的很简单。_websocket_ 创建一个 subject。这意味着你可以订阅它,通过调用 _next_ 来获得消息和发送消息。
|
||||
把 websocket 功能添加到你的应用中真的很简单。_websocket_ 创建一个 subject。这意味着你可以订阅它,通过调用 `next` 来获得消息和发送消息。
|
||||
|
||||
#### ajax
|
||||
|
||||
如你所知:类似于 websocket,提供 AJAX 查询的功能。你可能用了一个带有 AJAX 功能的库或者框架。或者你没有用,那么我建议使用 fetch(或者必要的话用 polyfill),把返回的 promise 封装到一个 observable 对象中(参考稍后会讲到的 _defer_ 函数)。
|
||||
如你所知:类似于 websocket,提供 AJAX 查询的功能。你可能用了一个带有 AJAX 功能的库或者框架。或者你没有用,那么我建议使用 fetch(或者必要的话用 polyfill),把返回的 promise 封装到一个 observable 对象中(参考稍后会讲到的 `defer` 函数)。
|
||||
|
||||
* * *
|
||||
|
||||
### Custom Streams
|
||||
### 定制流
|
||||
|
||||
有时候已有的函数用起来并不是足够灵活。或者你需要对订阅有更强的控制。
|
||||
|
||||
#### Subject
|
||||
#### 主题(`Subject`)
|
||||
|
||||
subject 是一个特殊的对象,它使得你的能够把数据发送到流中,并且能够控制数据。subject 本身就是一个 observable 对象,但如果你想要把流暴露给其它代码,建议你使用 _asObservable_ 方法。这样你就不能意外调用原始方法。
|
||||
`Subject` 是一个特殊的对象,它使得你的能够把数据发送到流中,并且能够控制数据。`Subject` 本身就是一个可观察对象,但如果你想要把流暴露给其它代码,建议你使用 `asObservable` 方法。这样你就不能意外调用原始方法。
|
||||
|
||||
```
|
||||
const subject = new Subject();
|
||||
const observable = subject.asObservable();
|
||||
```
|
||||
|
||||
```
|
||||
observable.subscribe();
|
||||
```
|
||||
|
||||
```
|
||||
subject.next(1);
|
||||
subject.next(2);
|
||||
subject.complete();
|
||||
@ -554,17 +476,11 @@ subject.complete();
|
||||
```
|
||||
const subject = new Subject();
|
||||
const observable = subject.asObservable();
|
||||
```
|
||||
|
||||
```
|
||||
subject.next(1);
|
||||
```
|
||||
|
||||
```
|
||||
observable.subscribe(console.log);
|
||||
```
|
||||
|
||||
```
|
||||
subject.next(2);
|
||||
subject.complete();
|
||||
```
|
||||
@ -574,20 +490,16 @@ subject.complete();
|
||||
// 2
|
||||
```
|
||||
|
||||
除了常规的 subject,RxJS 还提供了三种特殊的版本。
|
||||
除了常规的 `Subject`,RxJS 还提供了三种特殊的版本。
|
||||
|
||||
_AsyncSubject_ 在结束后只发送最后的一个值。
|
||||
`AsyncSubject` 在结束后只发送最后的一个值。
|
||||
|
||||
```
|
||||
const subject = new AsyncSubject();
|
||||
const observable = subject.asObservable();
|
||||
```
|
||||
|
||||
```
|
||||
observable.subscribe(console.log);
|
||||
```
|
||||
|
||||
```
|
||||
subject.next(1);
|
||||
subject.next(2);
|
||||
subject.complete();
|
||||
@ -598,18 +510,14 @@ subject.complete();
|
||||
// 2
|
||||
```
|
||||
|
||||
_BehaviorSubject_ 使得你能够提供一个(默认的)值,如果当前没有其它值发送的话,这个值会被发送给每个订阅者。否则订阅者收到最后一个发送的值。
|
||||
`BehaviorSubject` 使得你能够提供一个(默认的)值,如果当前没有其它值发送的话,这个值会被发送给每个订阅者。否则订阅者收到最后一个发送的值。
|
||||
|
||||
```
|
||||
const subject = new BehaviorSubject(1);
|
||||
const observable = subject.asObservable();
|
||||
```
|
||||
|
||||
```
|
||||
const subscription1 = observable.subscribe(console.log);
|
||||
```
|
||||
|
||||
```
|
||||
subject.next(2);
|
||||
subscription1.unsubscribe();
|
||||
```
|
||||
@ -622,29 +530,21 @@ subscription1.unsubscribe();
|
||||
|
||||
```
|
||||
const subscription2 = observable.subscribe(console.log);
|
||||
```
|
||||
|
||||
```
|
||||
// 输出
|
||||
// 2
|
||||
```
|
||||
|
||||
The _ReplaySubject_ 存储一定数量、或一定时间或所有的发送过的值。所有新的订阅者将会获得所有存储了的值。
|
||||
`ReplaySubject` 存储一定数量、或一定时间或所有的发送过的值。所有新的订阅者将会获得所有存储了的值。
|
||||
|
||||
```
|
||||
const subject = new ReplaySubject();
|
||||
const observable = subject.asObservable();
|
||||
```
|
||||
|
||||
```
|
||||
subject.next(1);
|
||||
```
|
||||
|
||||
```
|
||||
observable.subscribe(console.log);
|
||||
```
|
||||
|
||||
```
|
||||
subject.next(2);
|
||||
subject.complete();
|
||||
```
|
||||
@ -655,11 +555,11 @@ subject.complete();
|
||||
// 2
|
||||
```
|
||||
|
||||
你可以在 [ReactiveX documentation][10](它提供了一些其它的连接) 里面找到更多关于 subjects 的信息。[Ben Lesh][11] 在 [On The Subject Of Subjects][12] 上面提供了一些关于 subjects 的理解,[Nicholas Jamieson][13] 在 [in RxJS: Understanding Subjects][14] 上也提供了一些理解。
|
||||
你可以在 [ReactiveX 文档][10](它提供了一些其它的连接) 里面找到更多关于 `Subject` 的信息。[Ben Lesh][11] 在 [On The Subject Of Subjects][12] 上面提供了一些关于 `Subject` 的理解,[Nicholas Jamieson][13] 在 [in RxJS: Understanding Subjects][14] 上也提供了一些理解。
|
||||
|
||||
#### Observable
|
||||
#### 可观察对象
|
||||
|
||||
你可以简单地用 new 操作符创建一个 observable 对象。通过你传入的函数,你可以控制流,只要有人订阅了或者它接收到一个可以当成 subject 使用的 observer,这个函数就会被调用,比如,调用 next,complet 和 error。
|
||||
你可以简单地用 new 操作符创建一个可观察对象。通过你传入的函数,你可以控制流,只要有人订阅了或者它接收到一个可以当成 `Subject` 使用的观察者,这个函数就会被调用,比如,调用 `next`、`complet` 和 `error`。
|
||||
|
||||
让我们回顾一下列表示例:
|
||||
|
||||
@ -667,16 +567,12 @@ subject.complete();
|
||||
const myList = new List();
|
||||
myList.add(1);
|
||||
myList.add(3);
|
||||
```
|
||||
|
||||
```
|
||||
new Observable(observer => {
|
||||
for (let i = 0; i<list.size; i++) {
|
||||
observer.next(list.get(i));
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
observer.complete();
|
||||
})
|
||||
.subscribe();
|
||||
@ -687,14 +583,12 @@ new Observable(observer => {
|
||||
// 1 3 |
|
||||
```
|
||||
|
||||
这个函数可以返回一个 unsubcribe 函数,当有订阅者取消订阅时这个函数就会被调用。你可以用它来清楚或者执行一些收尾操作。
|
||||
这个函数可以返回一个 `unsubcribe` 函数,当有订阅者取消订阅时这个函数就会被调用。你可以用它来清楚或者执行一些收尾操作。
|
||||
|
||||
```
|
||||
new Observable(observer => {
|
||||
// 流式化
|
||||
```
|
||||
|
||||
```
|
||||
return () => {
|
||||
//clean up
|
||||
};
|
||||
@ -702,20 +596,18 @@ new Observable(observer => {
|
||||
.subscribe();
|
||||
```
|
||||
|
||||
#### 继承 Observable
|
||||
#### 继承可观察对象
|
||||
|
||||
在有可用的操作符前,这是一种实现自定义操作符的方式。RxJS 在内部扩展了 _Observable_。_Subject_ 就是一个例子,另一个是 _publisher_ 操作符。它返回一个 _ConnectableObservable_ 对象,该对象提供额外的方法 _connect_。
|
||||
在有可用的操作符前,这是一种实现自定义操作符的方式。RxJS 在内部扩展了 _可观察对象_ 。`Subject` 就是一个例子,另一个是 `publisher` 操作符。它返回一个 `ConnectableObservable` 对象,该对象提供额外的方法 `connect`。
|
||||
|
||||
#### 实现 Subscribable 接口
|
||||
#### 实现 `Subscribable` 接口
|
||||
|
||||
有时候你已经用一个对象来保存状态,并且能够发送值。如果你实现了 Subscribable 接口,你可以把它转换成一个 observable 对象。Subscribable 接口中只有一个 subscribe 方法。
|
||||
有时候你已经用一个对象来保存状态,并且能够发送值。如果你实现了 `Subscribable` 接口,你可以把它转换成一个可观察对象。`Subscribable` 接口中只有一个 `subscribe` 方法。
|
||||
|
||||
```
|
||||
interface Subscribable<T> { subscribe(observerOrNext?: PartialObserver<T> | ((value: T) => void), error?: (error: any) => void, complete?: () => void): Unsubscribable}
|
||||
```
|
||||
|
||||
* * *
|
||||
|
||||
### 结合和选择现有的流
|
||||
|
||||
知道怎么创建一个独立的流还不够。有时候你有好几个流但其实只需要一个。有些函数也可作为操作符,所以我不打算在这里深入展开。推荐看看 [Max NgWizard K][16] 所写的一篇 [文章][15],它还包含一些有趣的动画。
|
||||
@ -724,41 +616,34 @@ interface Subscribable<T> { subscribe(observerOrNext?: PartialObserver<T> | ((v
|
||||
|
||||
#### ObservableInput 类型
|
||||
|
||||
期望接收流的操作符和函数通常不单独和 observables 一起工作。相反,他们实际上期望的参数类型是 ObservableInput,定义如下:
|
||||
期望接收流的操作符和函数通常不单独和可观察对象一起工作。相反,它们实际上期望的参数类型是 ObservableInput,定义如下:
|
||||
|
||||
```
|
||||
type ObservableInput<T> = SubscribableOrPromise<T> | ArrayLike<T> | Iterable<T>;
|
||||
```
|
||||
|
||||
这意味着你可以传递一个 promises 或者数组却不需要事先把他们转换成 observables。
|
||||
这意味着你可以传递一个 promises 或者数组却不需要事先把他们转换成可观察对象。
|
||||
|
||||
#### defer
|
||||
|
||||
主要的目的是把一个 observable 对象的创建延迟(defer)到有人想要订阅的时间。在以下情况,这很有用:
|
||||
|
||||
* 创建 observable 对象的开销较大
|
||||
|
||||
* 你想要给每个订阅者新的 observable 对象
|
||||
|
||||
* 你想要在订阅时候选择不同的 observable 对象
|
||||
主要的目的是把一个 observable 对象的创建延迟(`defer`)到有人想要订阅的时间。在以下情况,这很有用:
|
||||
|
||||
* 创建可观察对象的开销较大
|
||||
* 你想要给每个订阅者新的可观察对象
|
||||
* 你想要在订阅时候选择不同的可观察对象
|
||||
* 有些代码必须在订阅之后执行
|
||||
|
||||
最后一点包含了一个并不起眼的用例:Promises(defer 也可以返回一个 promise 对象)。看看这个用到了 fetch API 的例子:
|
||||
最后一点包含了一个并不起眼的用例:Promises(`defer` 也可以返回一个 promise 对象)。看看这个用到了 fetch API 的例子:
|
||||
|
||||
```
|
||||
function getUser(id) {
|
||||
console.log("fetching data");
|
||||
return fetch(`https://server/user/${id}`);
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
const userPromise = getUser(1);
|
||||
console.log("I don't want that request now");
|
||||
```
|
||||
|
||||
```
|
||||
// 其它地方
|
||||
userPromise.then(response => console.log("done");
|
||||
```
|
||||
@ -770,17 +655,13 @@ userPromise.then(response => console.log("done");
|
||||
// done
|
||||
```
|
||||
|
||||
只要流在你订阅的时候执行了,promise 就会立即执行。我们调用 getUser 的瞬间,就发送了一个请求,哪怕我们这个时候不想发送请求。当然,我们可以使用 from 来把一个 promise 对象转换成 observable 对象,但我们传递的 promise 对象已经创建或执行了。defer 让我们能够等到订阅才发送这个请求:
|
||||
只要流在你订阅的时候执行了,promise 就会立即执行。我们调用 `getUser` 的瞬间,就发送了一个请求,哪怕我们这个时候不想发送请求。当然,我们可以使用 `from` 来把一个 promise 对象转换成可观察对象,但我们传递的 promise 对象已经创建或执行了。`defer` 让我们能够等到订阅才发送这个请求:
|
||||
|
||||
```
|
||||
const user$ = defer(() => getUser(1));
|
||||
```
|
||||
|
||||
```
|
||||
console.log("I don't want that request now");
|
||||
```
|
||||
|
||||
```
|
||||
// 其它地方
|
||||
user$.subscribe(response => console.log("done");
|
||||
```
|
||||
@ -794,7 +675,7 @@ user$.subscribe(response => console.log("done");
|
||||
|
||||
#### iif
|
||||
|
||||
_iif 包含了一个关于 _defer_ 的特殊用例:在订阅时选择两个流中的一个:
|
||||
`iif` 包含了一个关于 `defer` 的特殊用例:在订阅时选择两个流中的一个:
|
||||
|
||||
```
|
||||
iif(
|
||||
@ -810,9 +691,9 @@ iif(
|
||||
// AM before noon, PM afterwards
|
||||
```
|
||||
|
||||
引用了文档:
|
||||
引用该文档:
|
||||
|
||||
> 实际上 `[iif][3]` 能够轻松地用 `[defer][4]` 实现,它仅仅是出于方便和可读性的目的。
|
||||
> 实际上 [iif][3] 能够轻松地用 [defer][4] 实现,它仅仅是出于方便和可读性的目的。
|
||||
|
||||
#### onErrorResumeNext
|
||||
|
||||
@ -822,13 +703,9 @@ iif(
|
||||
const stream1$ = of(1, 2).pipe(
|
||||
tap(i => { if(i>1) throw 'error'}) //fail after first element
|
||||
);
|
||||
```
|
||||
|
||||
```
|
||||
const stream2$ = of(3,4);
|
||||
```
|
||||
|
||||
```
|
||||
onErrorResumeNext(stream1$, stream2$)
|
||||
.subscribe(console.log);
|
||||
```
|
||||
@ -848,9 +725,7 @@ onErrorResumeNext(stream1$, stream2$)
|
||||
function handleResponses([user, account]) {
|
||||
// 执行某些任务
|
||||
}
|
||||
```
|
||||
|
||||
```
|
||||
forkJoin(
|
||||
fetch("https://server/user/1"),
|
||||
fetch("https://server/account/1")
|
||||
@ -860,9 +735,9 @@ forkJoin(
|
||||
|
||||
#### merge / concat
|
||||
|
||||
发送每一个从源 observables 对象中发出的值。
|
||||
发送每一个从可观察对象源中发出的值。
|
||||
|
||||
_merge_ 接收一个参数,让你定义有多少流能被同时订阅。默认是无限制的。设为 1 就意味着监听一个源流,在它结束的时候订阅下一个。由于这是一个常见的场景,RxJS 为你提供了一个显示的函数:_concat_。
|
||||
`merge` 接收一个参数,让你定义有多少流能被同时订阅。默认是无限制的。设为 1 就意味着监听一个源流,在它结束的时候订阅下一个。由于这是一个常见的场景,RxJS 为你提供了一个显示的函数:`concat`。
|
||||
|
||||
```
|
||||
merge(
|
||||
@ -872,31 +747,20 @@ merge(
|
||||
2 //two concurrent streams
|
||||
)
|
||||
.subscribe();
|
||||
```
|
||||
|
||||
```
|
||||
// 只订阅流 1 和流 2
|
||||
```
|
||||
|
||||
```
|
||||
// 输出
|
||||
// Stream 1 -> after 1000ms
|
||||
// Stream 2 -> after 1200ms
|
||||
// Stream 1 -> after 2000ms
|
||||
```
|
||||
|
||||
```
|
||||
// 流 1 结束后,开始订阅流 3
|
||||
```
|
||||
|
||||
```
|
||||
// 输出
|
||||
// Stream 3 -> after 0 ms
|
||||
// Stream 2 -> after 400 ms (2400ms from beginning)
|
||||
// Stream 3 -> after 1000ms
|
||||
```
|
||||
|
||||
```
|
||||
|
||||
merge(
|
||||
interval(1000).pipe(mapTo("Stream 1"), take(2)),
|
||||
@ -908,9 +772,7 @@ concat(
|
||||
interval(1000).pipe(mapTo("Stream 1"), take(2)),
|
||||
interval(1200).pipe(mapTo("Stream 2"), take(2))
|
||||
)
|
||||
```
|
||||
|
||||
```
|
||||
// 输出
|
||||
// Stream 1 -> after 1000ms
|
||||
// Stream 1 -> after 2000ms
|
||||
@ -920,7 +782,7 @@ concat(
|
||||
|
||||
#### zip / combineLatest
|
||||
|
||||
_merge_ 和 _concat_ 一个接一个的发送所有从源流中读到的值,而 zip 和 combineLatest 是把每个流中的一个值结合起来一起发送。_zip_ 结合所有源流中发送的第一个值。如果流的内容相关联,那么这就很有用。
|
||||
`merge` 和 `concat` 一个接一个的发送所有从源流中读到的值,而 `zip` 和 `combineLatest` 是把每个流中的一个值结合起来一起发送。`zip` 结合所有源流中发送的第一个值。如果流的内容相关联,那么这就很有用。
|
||||
|
||||
```
|
||||
zip(
|
||||
@ -935,7 +797,7 @@ zip(
|
||||
// [0, 0] [1, 1] [2, 2] ...
|
||||
```
|
||||
|
||||
_combineLatest_ 与之类似,但结合的是源流中发送的最后一个值。直到所有源流至少发送一个值之后才会触发事件。这之后每次源流发送一个值,它都会把这个值与其他流发送的最后一个值结合起来。
|
||||
`combineLatest` 与之类似,但结合的是源流中发送的最后一个值。直到所有源流至少发送一个值之后才会触发事件。这之后每次源流发送一个值,它都会把这个值与其他流发送的最后一个值结合起来。
|
||||
|
||||
```
|
||||
combineLatest(
|
||||
@ -983,15 +845,13 @@ race(
|
||||
// foo |
|
||||
```
|
||||
|
||||
由于 _of_ 立即产生一个值,因此它是最快的流,然而这个流就被选中了。
|
||||
|
||||
* * *
|
||||
由于 `of` 立即产生一个值,因此它是最快的流,然而这个流就被选中了。
|
||||
|
||||
### 总结
|
||||
|
||||
已经有很多创建 observables 对象的方式了。如果你想要创造响应式的 APIs 或者想用响应式的 API 结合传统 APIs,那么了解这些方法很重要。
|
||||
已经有很多创建可观察对象的方式了。如果你想要创造响应式的 API 或者想用响应式的 API 结合传统 API,那么了解这些方法很重要。
|
||||
|
||||
我已经向你展示了所有可用的方法,但它们其实还有很多内容可以讲。如果你想更加深入地了解,我极力推荐你查阅 [documentation][20] 或者阅读相关文章。
|
||||
我已经向你展示了所有可用的方法,但它们其实还有很多内容可以讲。如果你想更加深入地了解,我极力推荐你查阅 [文档][20] 或者阅读相关文章。
|
||||
|
||||
[RxViz][21] 是另一种值得了解的有意思的方式。你编写 RxJS 代码,产生的流可以用图形或动画进行显示。
|
||||
|
||||
@ -1001,7 +861,7 @@ via: https://blog.angularindepth.com/the-extensive-guide-to-creating-streams-in-
|
||||
|
||||
作者:[Oliver Flaggl][a]
|
||||
译者:[BriFuture](https://github.com/BriFuture)
|
||||
校对:[校对者ID](https://github.com/校对者ID)
|
||||
校对:[wxy](https://github.com/wxy)
|
||||
|
||||
本文由 [LCTT](https://github.com/LCTT/TranslateProject) 原创编译,[Linux中国](https://linux.cn/) 荣誉推出
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user