专栏首页草根专栏RxJS速成 (下)

RxJS速成 (下)

Subject

Subject比较特殊, 它即是Observable又是Observer.

作为Observable, Subject是比较特殊的, 它可以对多个Observer进行广播, 而普通的Observable只能单播, 它有点像EventEmitters(事件发射器), 维护着多个注册的Listeners.

作为Observable, 你可以去订阅它, 提供一个Observer就会正常的收到推送的值. 从Observer的角度是无法分辨出这个Observable是单播的还是一个Subject.

从Subject内部来讲, subscribe动作并没有调用一个新的执行来传递值, 它只是把Observer注册到一个列表里, 就像其他库的AddListener一样.

作为Observer, 它是一个拥有next(), error(), complete()方法的对象, 调用next(value)就会为Subject提供一个新的值, 然后就会多播到注册到这个Subject的Observers.

例子 subject.ts:

import { Subject } from "rxjs/Subject";

const subject = new Subject();

const subscriber1 = subject.subscribe({
    next: (v) => console.log(`observer1: ${v}`)
});
const subscriber2 = subject.subscribe({
    next: (v) => console.log(`observer2: ${v}`)
});

subject.next(1);
subscriber2.unsubscribe();
subject.next(2);

const subscriber3 = subject.subscribe({
    next: (v) => console.log(`observer3: ${v}`)
});

subject.next(3);

订阅者1,2从开始就订阅了subject. 然后subject推送值1的时候, 它们都收到了. 

然后订阅者2, 取消了订阅, 随后subject推送值2, 只有订阅者1收到了.

后来订阅者3也订阅了subject, 然后subject推送了3, 订阅者1,3都收到了这个值.

下面是一个angular 5的例子:

app.component.html:

<h3>从Subject共享Observable到多个Subscribers</h3>
<input type="text" placeholder="start typing" (input)="mySubject.next($event)" (keyup)="mySubject.next($event)">

<br> Subscriber to input events got {{inputValue}}
<br>
<br> Subscriber to keyup events got {{keyValue}}

app.component.ts:

import { Component } from '@angular/core';
import { Subject } from 'rxjs/Subject';
import 'rxjs/add/operator/filter';
import 'rxjs/add/operator/map';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent {
  title = 'app';

  keyValue: string;
  inputValue: string;

  mySubject: Subject<Event> = new Subject();

  constructor() {
    // subscriber 1
    this.mySubject.filter(({ type }) => type === 'keyup')
      .map(e => (<KeyboardEvent>e).key)
      .subscribe(value => this.keyValue = value);

    // subscriber 2
    this.mySubject.filter(({ type }) => type === 'input')
      .map(e => (<HTMLInputElement>e.target).value)
      .subscribe(value => this.inputValue = value);
  }
}

input和keyup动作都把event推送到mySubject, 然后mySubject把值推送给订阅者, 订阅者1通过过滤和映射它只处理keyup类型的事件, 而订阅者2只处理input事件.

效果:

BehaviorSubject

BehaviorSubject 是Subject的一个变种, 它有一个当前值的概念, 它会把它上一次发送给订阅者值保存起来, 一旦有新的Observer进行了订阅, 那这个Observer马上就会从BehaviorSubject收到这个当前值.

也可以这样理解BehaviorSubject的特点:

  • 它代表一个随时间变化的值, 例如, 生日的流就是Subject, 而一个人的年龄流就是BehaviorSubject.
  • 每个订阅者都会从BehaviorSubject那里得到它推送出来的初始值和最新的值.
  • 用例: 共享app状态.

例子 behavior-subject.ts:

import { BehaviorSubject } from "rxjs/BehaviorSubject";

const subject = new BehaviorSubject(0);

subject.subscribe({
    next: v => console.log(`Observer1: ${v}`)
});

subject.next(1);
subject.next(2);

subject.subscribe({
    next: v => console.log(`Observer2: ${v}`)
});

subject.next(3);

效果:

常用Operators:

concat 

concat: 按顺序合并observables. 只会在前一个observable结束之后才会订阅下一个observable.

它适合用于顺序处理, 例如http请求.

例子: 

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/timer';
import 'rxjs/add/operator/mapTo';
import 'rxjs/add/observable/concat';

let firstReq = Observable.timer(3000).mapTo('First Response');
let secondReq = Observable.timer(1000).mapTo('Second Response');

Observable.concat(firstReq, secondReq)
    .subscribe(res => console.log(res));

效果:

merge

把多个输入的observable交错的混合成一个observable, 不按顺序.

merge实际上是订阅了每个输入的observable, 它只是把输入的observable的值不带任何转换的发送给输出的Observable. 只有当所有输入的observable都结束了, 输出的observable才会结束. 任何在输入observable传递来的错误都会立即发射到输出的observable, 也就是把整个流都杀死了 .

例子:

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/timer';
import 'rxjs/add/operator/mapTo';
import 'rxjs/add/observable/merge';

let firstReq = Observable.timer(3000).mapTo('First Response');
let secondReq = Observable.timer(1000).mapTo('Second Response');

Observable.merge(firstReq, secondReq)
    .subscribe(res => console.log(res));

效果:

mergeMap (原来叫flatMap)

mergeMap把每个输入的Observable的值映射成Observable, 然后把它们混合成一个Observable.

mergeMap可以把嵌套的observables拼合成非嵌套的observable.

它有这些好处:

  • 不必编写嵌套的subscribe()
  • 把每个observable发出来的值转换成另一个observable
  • 自动订阅内部的observable并且把它们(可能)交错的合成一排.

这个还是通过例子来理解比较好:

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/from';
import 'rxjs/add/operator/mergeMap';

function getData() {
    const students = Observable.from([
        { name: 'Dave', age: 17 },
        { name: 'Nick', age: 18 },
        { name: 'Lee', age: 15 }
    ]);

    const teachers = Observable.from([
        { name: 'Miss Wan', age: 28 },
        { name: 'Mrs Wang', age: 31 },
    ]);

    return Observable.create(
        observer => {
            observer.next(students);
            observer.next(teachers);
        }
    );
}

getData()
    .mergeMap(persons => persons)
    .subscribe(
        p => console.log(`Subscriber got ${p.name} - ${p.age}`)
    );

效果:

switchMap

switchMap把每个值都映射成Observable, 然后使用switch把这些内部的Observables合并成一个.

switchMap有一部分很想mergeMap, 但也仅仅是一部分像而已.

因为它还具有取消的效果, 每次发射的时候, 前一个内部的observable会被取消, 下一个observable会被订阅. 可以把这个理解为切换到一个新的observable上了.

这个还是看marble图比较好理解:

例子: 

// 立即发出值, 然后每5秒发出值
const source = Rx.Observable.timer(0, 5000);
// 当 source 发出值时切换到新的内部 observable,发出新的内部 observable 所发出的值
const example = source.switchMap(() => Rx.Observable.interval(500));
// 输出: 0,1,2,3,4,5,6,7,8,9...0,1,2,3,4,5,6,7,8
const subscribe = example.subscribe(val => console.log(val));

更好的例子是: 网速比较慢的时候, 客户端发送了多次重复的请求, 如果前一次请求在2秒内没有返回的话, 那么就取消前一次请求, 不再需要前一次请求的结果了, 这里就应该使用debounceTime配合switchMap.

mergeMap vs switchMap的例子

mergeMap:

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/interval';
import 'rxjs/add/operator/take';
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/mergeMap';
import 'rxjs/add/operator/switchMap';

const outer = Observable.interval(1000).take(2);

const combined = outer.mergeMap(x => {
    return Observable.interval(400)
        .take(3)
        .map(y => `outer ${x}: inner ${y}`);
});

combined.subscribe(res => console.log(`result ${res}`));

效果:

switchMap:

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/interval';
import 'rxjs/add/operator/take';
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/mergeMap';
import 'rxjs/add/operator/switchMap';

const outer = Observable.interval(1000).take(2);

const combined = outer.switchMap(x => {
    return Observable.interval(400)
        .take(3)
        .map(y => `outer ${x}: inner ${y}`);
});

combined.subscribe(res => console.log(`result ${res}`));

zip

zip操作符也会合并多个输入的observables成为一个observable. 多个输入的observable的值, 按顺序, 按索引进行合并, 如果某一个observable在该索引上的值还没有发射值, 那么会等它, 直到所有的输入observables在该索引位置上的值都发射出来, 输出的observable才会发射该索引的值.

例子:

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/of';
import 'rxjs/add/observable/zip';

let age$ = Observable.of<number>(27, 25, 29);
let name$ = Observable.of<string>('Foo', 'Bar', 'Beer');
let isDev$ = Observable.of<boolean>(true, true, false);

Observable
    .zip(age$,
        name$,
        isDev$,
        (age: number, name: string, isDev: boolean) => ({ age, name, isDev }))
    .subscribe(x => console.log(x));

效果:

就不往下写了, 其实看文档就行, 最重要的还是上一部分.

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 使用Identity Server 4建立Authorization Server (6) - js(angular5) 客户端

    由于手头目前用项目, 所以与前几篇文章不同, 这次要讲的js客户端这部分是通过我刚刚开发的真是项目的代码来讲解的. 这是后端的代码: https://githu...

    solenovex
  • RxJS速成

    RxJS是ReactiveX编程理念的JavaScript版本。ReactiveX是一种针对异步数据流的编程。简单来说,它将一切数据,包括HTTP请求,DOM事...

    solenovex
  • gRPC in ASP.NET Core 3.x -- Protocol Buffer(2)Go语言的例子(下)

    第一篇文章(大约半年前写的):https://www.cnblogs.com/cgzl/p/11246324.html

    solenovex
  • SpringBoot集成Swagger2

    在一些接口项目中,API的使用很频繁,所以一款API在线文档生成和测试工具非常有必要。而Swagger UI就是这么一款很实用的在线工具 本博客介绍如何在公司...

    SmileNicky
  • Spring Boot系列——7步集成RabbitMQ

    RabbitMQ是一种我们经常使用的消息中间件,通过RabbitMQ可以帮助我们实现异步、削峰的目的。

    JackieZheng
  • Java中使用线程时,请不要忘记Spring TaskExecutor组件

    当我们实现的web应用程序需要长时间运行一个任务时,Spring TaskExecutor管理组件是一个很好选择,会给我们代码的实现提供很大的方便,也会节省时间...

    用户1289394
  • golang 基于Mac os 构建镜像

    自定义基础镜像官方文档看完后最还是google一下如何构建,tar cv --files-from /dev/null | docker import - sc...

    笨兔儿
  • Android中使用ShareSDK集成分享功能的实例代码

    现在APP开发集成分享功能已经是非常普遍的需求了。其他集成分享技术我没有使用过,今天我就来介绍下使用ShareSDK来进行分享功能开发的一些基本步骤和注意点,帮...

    砸漏
  • javascript面向对象之继承(上)

    我们之前介绍了javascript面向对象的封装的相关内容,还介绍了js的call方法,今天开始讨论js的继承 这篇文章参考了《javascript高级程序设计...

    陌上寒
  • 爬虫(104)教你词云分析拉勾网数百个职位招聘详

    昨天我们分析了某 girl 的 QQ 空间,之后想想还是不过瘾啊,感觉还可以深度挖掘词云这个库,于是在网上找了一个实际例子又来波

    公众号---志学Python

扫码关注云+社区

领取腾讯云代金券