【响应式编程的思维艺术】 （5）Angular中Rxjs的应用示例

摘要：本文是响应式编程第四章构建完整的应用程序这篇文章的学习笔记。涉及的运算符每隔指定时间将流中的数据以数组形式推送出去。中提供了一种叫做异步管道的模板语法，可以直接在的微语法中使用可观测对象示例五一点建议一定要好好读官方文档。

本文是【Rxjs 响应式编程-第四章 构建完整的Web应用程序】这篇文章的学习笔记。
示例代码托管在：http://www.github.com/dashnowords/blogs
博客园地址：《大史住在大前端》原创博文目录
华为云社区地址：【你要的前端打怪升级指南】

[TOC]

RxJS-DOM

原文示例中使用这个库进行DOM操作，笔者看了一下github仓库，400多星，而且相关的资料很少，所以建议理解思路即可，至于生产环境的使用还是三思吧。开发中Rxjs几乎默认是和Angular技术栈绑定在一起的，笔者最近正在使用ionic3进行开发，本篇将对基本使用方法进行演示。

冷热Observable

冷Observable从被订阅时就发出整个值序列

热Observable无论是否被订阅都会发出值，机制类似于javascript事件。

涉及的运算符

bufferWithTime(time:number)-每隔指定时间将流中的数据以数组形式推送出去。

pluck(prop:string)- 操作符，提取对象属性值，是一个柯里化后的函数，只接受一个参数。

Angular应用中基本HTTP请求的方式:

import { Injectable } from "@angular/core";
import { Observable, of } from "rxjs";
import { MessageService } from "./message.service";//某个自定义的服务
import { HttpClient, HttpParams, HttpResponse } from "@angular/common/http";

@Injectable({
  providedIn: "root"
})
export class HeroService {
  private localhost = "http://localhost:3001";
  private all_hero_api = this.localhost + "/hero/all";//查询所有英雄
  private query_hero_api = this.localhost + "/hero/query";//查询指定英雄

  constructor(private http:HttpClient) {
  }
  
  /*一般get请求*/
  getHeroes(): Observable>{
    return this.http.get(this.all_hero_api,{observe:"response"});
  }

  /*带参数的get请求*/
  getHero(id: number): Observable>{
    let params = new HttpParams();
        params.set("id", id+"");
        return this.http.get(this.query_hero_api,{params:params,observe:"response"});
  }
  
  /*带请求体的post请求，any可以自定义响应体格式*/
  createHero(newhero: object): Observable>{
      return this.http.post>(this.create_hero_api,{data:newhero},{observe:"response"});
  } 
}

在express中写一些用于测试的虚拟数据:

var express = require("express");
var router = express.Router();

/* GET home page. */
router.get("/all", function(req, res, next) {
  let heroes = [{
    index:1,
    name:"Thor",
    hero:"God of Thunder"
  },{
    index:2,
    name:"Tony",
    hero:"Iron Man"
  },{
    index:3,
    name:"Natasha",
    hero:"Black Widow"
  }]
  res.send({
     data:heroes,
     result:true
  })
});

/* GET home page. */
router.get("/query", function(req, res, next) {
  console.log(req.query);
  let hero= {
    index:4,
    name:"Steve",
    hero:"Captain America"
  }
  res.send({
     data:hero,
     result:true
  })
});


/* GET home page. */
router.post("/create", function(req, res, next) {
  console.log(req.body);
  let newhero = {
     index:5,
     name:req.body.name,
     hero:"New Hero"
  }
  res.send({
     data:newhero,
     result:true
  })
});

module.exports = router;

在组件中调用上面定义的方法：

sendGet(){
 this.heroService.getHeroes().subscribe(resp=>{
   console.log("响应信息:",resp);
   console.log("响应体:",resp.body["data"]);
 })
}

sendQuery(){
this.heroService.getHero(1).subscribe(resp=>{
  console.log("响应信息:",resp);
  console.log("响应体:",resp.body["data"]);
})
}

sendPost(){
this.heroService.createHero({name:"Dash"}).subscribe(resp=>{
  console.log("响应信息:",resp);
  console.log("响应体:",resp.body["data"]);
})
}

控制台打印的信息可以看到后台的虚拟数据已经被请求到了：

尽管看起来Http请求的返回结果是一个可观测对象，但是它却没有map方法，当需要对http请求返回的可观测对象进行操作时，可以使用pipe操作符来实现：

import { Observable, of, from} from "rxjs";
import { map , tap, filter, flatMap }from "rxjs/operators";

/*构建一个模拟的结果处理管道
*map操作来获取数据
*tap实现日志
*flatMap实现结果自动遍历
*filter实现结果过滤
*/
getHeroes$(): Observable>{
    return this.http.get(this.all_hero_api,{observe:"response"})
    .pipe(
          map(resp=>resp.body["data"]),
          tap(this.log),
          flatMap((data)=>{return from(data)}),
          filter((data)=>data["index"] > 1)
    );
}

很熟悉吧？经过处理管道后，一次响应中的结果数据被转换为逐个发出的数据，并过滤掉了不符合条件的项：

Angular中文网列举了最常用的一些操作符,RxJS官方文档有非常详细的示例及说明，且均配有形象的大理石图，建议先整体浏览一下有个印象，有需要的读者可以每天熟悉几个，很快就能上手，运算符的使用稍显抽象，且不同运算符的组合使用在流程控制和数据处理方面的用法灵活多变，也是有很多套路的，开发经验需要慢慢积累。

原文中提到的冷热Observable的差别可以参考这篇文章【RxJS：冷热模式的比较】，概念本身并不难理解。

开发中常会遇到这样一种场景，某些集合型的常量，完全是可以复用的，通常开发者会将其进行缓存至某个全局单例中，接着在优化阶段，通过增加一个if判断在请求之前先检查缓存再决定是否需要请求，Rxjs提供了一种更优雅的实现。

先回顾一下上面的http请求代码:

getHeroes(): Observable>{
   return this.http.get(this.all_hero_api,{observe:"response"});
}

http请求默认返回一个冷Observable，每当返回的流被订阅时就会触发一个新的http请求,Rxjs中通过shareReplay( )操作符将一个可观测对象转换为热Observable（注意：shareReplay( )不是唯一一种可以加热Observable的方法），这样在第一次被订阅时，网络请求被发出并进行了缓存，之后再有其他订阅者加入时，就会得到之前缓存的数据，运算符的名称已经很清晰了，【share-共享】，【replay-重播】，是不是形象又好记。对上面的流进行一下转换：

  getHeroes$(): Observable>{
    return this.http.get(this.all_hero_api,{observe:"response"})
    .pipe(
      map(resp=>resp.body["data"]),
      tap(this.log),
      flatMap((data)=>{return from(data)}),
      filter((data)=>data["index"] > 1),
      shareReplay() // 转换管道的最后将这个流转换为一个热Observable
    )
  }

在调用的地方编写调用代码：

sendGet(){
     let obs = this.heroService.getHeroes$();
     //第一次被订阅
     obs.subscribe(resp=>{
       console.log("响应信息:",resp);
     });
    //第二次被订阅
     setTimeout(()=>{
       obs.subscribe((resp)=>{
         console.log("延迟后的响应信息",resp);
       })
     },2000)
}

通过结果可以看出，第二次订阅没有触发网络请求，但是也得到了数据：

网络请求只发送了一次（之前的会发送两次）：

这种场景笔者并没有进行生产实践，一是因为这种模式需要将数据的变换处理全部通过pipe( )管道来进行，笔者自己的函数式编程功底可能还不足以应付，二来总觉得很多示例的使用场景很牵强，所以仅作基本功能介绍，后续有实战心得后再修订补充。Angular中提供了一种叫做异步管道的模板语法，可以直接在*ngFor的微语法中使用可观测对象：

  
{{contact.name}}


  
{{contact.name}}

示例：

this.contacts = http.get("contacts.json")
                    .map(response => response.json().items)
                    .share();
setTimeout(() => this.contacts2 = this.contacts, 500);

一定要好好读官方文档。