【响应式编程的思维艺术】 （3）flatMap背后的代数理论Monad

目录

• 一. 划重点
• 二. flatmap功能解析
• 三. flatmap的推演
• 四. 资料参考

本文是rxjs 响应式编程-第二章：序列的深入研究这篇文章的学习笔记。

示例代码托管在：

更多博文：

一. 划重点

文中使用到的一些基本运算符：

• map-映射
• filter-过滤
• reduce-有限列聚合
• scan-无限列聚合
• flatmap-拉平操作（重点）
• catch-捕获错误
• retry-序列重试
• from-生成可观测序列
• range-生成有限的可观测序列
• interval-每隔指定时间发出一次顺序整数
• distinct-去除出现过的重复值

建议自己动手尝试一下，记住就可以了，有过lodash使用经验的开发者来说并不难。

二. flatmap功能解析

原文中在http请求拿到获取到数据后，最初使用了foreach实现了手动流程管理，于是原文提出了优化设想，试图探究如何依赖响应式编程的特性将手动的数据加工转换改造为对流的转换，好让最终的消费者能够拿到直接可用的数据，而不是得到一个响应后手动进行很多后处理。在代码层面需要解决的问题就是，如何在不使用手动遍历的前提下将一个有限序列中的数据逐个发给订阅者，而不是一次性将整个数据集发过去。

假设我们现在并不知道有flatmap这样一个可以使用的方法，那么先来做一些尝试：

var quakes = rx.observable.create(function(observer) {
    //模拟得到的响应流
    var response = { 
       features:[{
        earth:1
       },{
        earth:2
       }],
       test:1
    }
/*  最初的手动遍历代码  
    var quakes = response.features;
        quakes.foreach(function(quake) {
            observer.onnext(quake);
        });*/

    observer.onnext(response);
})
//为了能将features数组中的元素逐个发送给订阅者，需要构建新的流
.map(dataset){
    return rx.observable.from(dataset.features)
}

当我们订阅quakes这个事件流的时候，每次都会得到另一个observable，它是因为数据源经过了映射变换，从数据变成了可观测对象。那么为了得到最终的序列值，就需要再次订阅这个observable，这里需要注意的是可观测对象被订阅前是不启动的，所以不用担心它的时序问题。

quakes.subscribe(function(data){
    data.subscribe(function(quake){
        console.log(quake);
    })
})；

如果将observable看成一个盒子，那么每一层盒子只是实现了流程控制功能性的封装，为了取得真正需要使用的数据，最终的订阅者不得不像剥洋葱似的通过subscribe一层层打开盒子拿到最里面的数据，这样的封装性对于数据在流中的传递具有很好的隔离性，但是对最终的数据消费者而言，却是一件很麻烦的事情。

这时flatmap运算符就派上用场了，它可以将冗余的包裹除掉，从而在主流被订阅时直接拿到要使用的数据，从大理石图来直观感受一下flatmap:

乍看之下会觉得它和merge好像是一样的，其实还是有一些区别的。merge的作用是将多个不同的流合并成为一个流，而上图中a1，a2，a3这三个流都是当主流a返回数据时新生成的，可以将他们想象为a的支流，如果你想在支流里捞鱼，就需要在每个支流里布网，而flatmap相当于提供了一张大网，将所有a的支流里的鱼都给捞上来。

所以在使用了flatmap后，就可以直接在一级订阅中拿到需要的数据了：

var quakes = rx.observable.create(function(observer) {
    var response = { 
       features:[{
        earth:1
       },{
        earth:2
       }],
       test:1
    }
    observer.onnext(response);
}).flatmap((data)=>{
    return rx.observable.from(data.features);
});

quakes.subscribe(function(quake) {
   console.log(quake)
});

三. flatmap的推演

3.1 函数式编程基础知识回顾

如果本节的基本知识你尚不熟悉，可以通过javascript基础修炼（8）——指向fp世界的箭头函数这篇文章来简单回顾一下函数式编程的基本知识，然后再继续后续的部分。

/*map运算符的作用
*对所有容器类而言，它相当于打开容器，进行操作，然后把容器再盖上。
*container在这里只是一个抽象定义，为了看清楚它对于容器中包含的值意味着什么。
*你会发现它其实就是observable的抽象原型。
*/
container.prototype.map = function(f){
  return container.of(f(this.__value))
}

//基本的科里化函数
var curry = function(fn){
  args = [].slice.call(arguments, 1);
  return function(){
     [].push.apply(args, arguments);
     return fn.apply(this, args);
  }
}

//map pointfree风格的map运算符
var map = curry(function(f, any_functor_at_all) {
  return any_functor_at_all.map(f);
});

/*compose函数组合方法
*运行后返回一个新函数，这个函数接受一个参数。
*函数科里化的基本应用，也是函数式编程中运算管道构建的基本方法。
*/
var compose = function (f, g) {
    return function (x) {
        return f(g(x));
    }
};
/*io容器
*一个简单的container实现，用来做流程管理
*这里需要注意，io实现的作用是函数的缓存，且总是返回新的io实例
*可以看做一个简化的promise,重点是直观感受一下它作为函数的
*容器是如何被使用的，对于理解observable有很大帮助
*/
var io = function(f) {
  this.__value = f;
}

io.of = function(x) {
  return new io(function() {
    return x;
  });
}

io.prototype.map = function(f) {
  return new io(compose(f, this.__value));
}

如果上面的基本知识没有问题，那么就继续。

3.2 从一个容器的例子开始

现在来实现这样一个功能，读入一个文件的内容，将其中的a字符全部换成b字符，接着存入另一个文件，完成后在控制台输出一个消息，为了更明显地看到数据容器的作用，我们使用同步方法并将其包裹在io容器中，然后利用函数式编程:

var fs = require('fs');

//读取文件
var readfile = (filename)=>io.of(fs.readfilesync(filename,'utf-8'));

//转换字符
var transcontent = (content)=>io.of((content)=>content.replace('a','b'));

//写入字符串
var writefile = (content)=>io.of(fs.writefilesync('dest.txt',content));

当具体的函数被io容器包裹起来而实现延迟执行的效果时，就无法按原来的方式使用compose( )运算符直接对功能进行组合，因为readfile函数运行时的输出结果（一个io容器实例）和transcontent函数需要的参数类型（字符串）不再匹配，在不修改原有函数定义的前提下，函数式编程中采用的做法是使用map操作符来预置一个参数：

/*
*map(transcontent)是一个高阶函数，它的返回函数就可以接收一个容器实例，
*并对容器中的内容执行map操作。
*/
var taskstep12 = compose(map(transcontent), readfile);

这里比较晦涩，涉及到很多功能性函数的嵌套，建议手动推导一下taskstep12这个变量的值，它的结构是这样一种形式：

io{
    __value:io{
        __value:somecomposedfnexpression
    }
}

如果试图一次性将所有的步骤组合在一起，就需要采用下面的形式：

var task = compose(map(map(writefile)),map(transcontent),readfile);
//组合后的task形式就是
//io{io{io{__value:somecomposedfnexpression}}}

问题已经浮出水面了，每多加一个针对容器操作的步骤，书写时就需要多包裹一层map，而运行时就需要多进入一层才能触及组合好的可以实现真正功能的函数表达式，真的是很麻烦。