angular-rxjs

rxjs和promise比较像，也是解决异步数据流的一种方案，不过rxjs比promise的功能更强大，可以理解成升级版的promise。
下面先来看一下基本使用：

import { Component, OnInit } from '@angular/core';
import { Observable } from 'rxjs';//rxjs已经集成到angular
@Component({
  selector: 'app-rxjs',
  templateUrl: './rxjs.component.html',
  styleUrls: ['./rxjs.component.scss']
})
export class RxjsComponent implements OnInit {

  constructor() { }

  ngOnInit(): void {
    this.getPromiseData().then(res=>{
      console.log(res)
    })
    this.getRxjsData().subscribe(res=>{
      console.log(res)
    })
  }
  getPromiseData() {
    return new Promise((resolve) => {
      setTimeout(() => {
        let listdata = [{ id: 1, name: 'a' }, { id: 2, name: 'b' }];
        resolve(listdata);
      }, 1000)
    })

  }

  getRxjsData() {

    let stream = new Observable(observer => {
      setTimeout(() => {
        let listdata = [{ id: 6, name: 'c' }, { id: 7, name: 'd' }];
        observer.next(listdata)
      }, 2000)
    })
    return stream;

  }

}

使用上看没啥区别，无非换了几个关键字：
promise的resolve换成了next，而then换成了subscribe。
但实际上rxjs更多promise不具备的功能。

取消订阅

假如某一个异步数据接口的请求时间不稳定，如果等待时间过长，我们希望取消本次请求，改成其他的方案，怎么办呢？
如果是promise，那我们只能被动等待resolve的发生而无法取消本次请求，而rxjs则可以取消本次请求的订阅。

import { Component, OnInit } from '@angular/core';
import { Observable } from 'rxjs';//rxjs已经集成到angular
@Component({
  selector: 'app-rxjs',
  templateUrl: './rxjs.component.html',
  styleUrls: ['./rxjs.component.scss']
})
export class RxjsComponent implements OnInit {

  constructor() { }

  ngOnInit(): void {
    this.getPromiseData().then(res=>{
      console.log(res)
    })
    
    let subscribeEvent = this.getRxjsData().subscribe(res=>{
      console.log(res)
    }) 
   
   setTimeout(()=>{// 虽然我订阅了异步数据，但如果等待时间超过了一秒钟我就取消订阅
    subscribeEvent.unsubscribe()
   },1000) 
  }
  getPromiseData() {
    return new Promise((resolve) => {
      setTimeout(() => {
        let listdata = [{ id: 1, name: 'a' }, { id: 2, name: 'b' }];
        resolve(listdata);
      }, 1000)
    })

  }

  getRxjsData() {

    let stream = new Observable(observer => {
      setTimeout(() => {
        let listdata = [{ id: 6, name: 'c' }, { id: 7, name: 'd' }];
        observer.next(listdata)
      }, 2000) // 2s后发起
    })
    return stream;

  }

}

订阅后多次执行

假如你写了一个爬虫程序，需要对一个网站频繁发起请求，但爬取一段时间后，对方网站必然会屏蔽掉你的请求，这时我们希望能够一段时间后自动再次发起请求，怎么实现呢？
如果是promise的话，setInterval里面执行resolve只能执行一次，如果想多次执行的话，就只能靠rxjs了。

import { Component, OnInit } from '@angular/core';
import { Observable } from 'rxjs';//rxjs已经集成到angular
@Component({
  selector: 'app-rxjs',
  templateUrl: './rxjs.component.html',
  styleUrls: ['./rxjs.component.scss']
})
export class RxjsComponent implements OnInit {

  constructor() { }

  ngOnInit(): void {
    this.getPromiseData().then(res=>{
      console.log(res)
    })
    
    this.getRxjsData().subscribe(res=>{
      console.log(res)
    }) 
   

  }
  getPromiseData() {
    return new Promise((resolve) => {
      setInterval(() => {
        let listdata = [{ id: 1, name: 'a' }, { id: 2, name: 'b' }];
        resolve(listdata);
      }, 1000)
    })

  }

  getRxjsData() {

    let stream = new Observable(observer => {
      setInterval(() => {
        let listdata = [{ id: 6, name: 'c' }, { id: 7, name: 'd' }];
        observer.next(listdata)
      }, 2000) // 2s后发起
    })
    return stream;

  }

}

操作符

在拿到我们订阅的异步数据之前，我们也可以对数据进行过滤和加工等操作，这些操作需要用管道将这些操作符串联起来：

import { Component, OnInit } from '@angular/core';
import { Observable } from 'rxjs';//rxjs已经集成到angular
import { filter, map } from 'rxjs/operators';
@Component({
  selector: 'app-rxjs',
  templateUrl: './rxjs.component.html',
  styleUrls: ['./rxjs.component.scss']
})
export class RxjsComponent implements OnInit {

  constructor() { }

  ngOnInit(): void {

    this.getRxjsData().pipe(

      map((value:string)=>{ //将字符串转换为json
        
        return JSON.parse(value)
      })
      
      ).subscribe(res=>{
      console.log(typeof res)
    }) 
   

  }

  getRxjsData() {

    let stream = new Observable(observer => {
      setTimeout(() => {
        let str = JSON.stringify({code:10000,data:[{ id: 6, name: 'c' }, { id: 7, name: 'd' }]}); //返回的是字符串
        observer.next(str)
      }, 2000) // 2s后发起
    })
    return stream;

  }

}