✨从响应式讲起，Observable：穿个马甲你就不认识啦？（附实战）

掘金安东尼

发布于 2022-11-16 16:36:29

1.1K0

发布于 2022-11-16 16:36:29

本文为稀土掘金技术社区首发签约文章，14天内禁止转载，14天后未获授权禁止转载，侵权必究!

专栏简介

作为一名 5 年经验的 JavaScript 技能拥有者，笔者时常在想，它的核心是什么？后来我确信答案是：闭包和异步。而函数式编程能完美串联了这两大核心，从高阶函数到函数组合；从无副作用到延迟处理；从函数响应式到事件流，从命令式风格到代码重用。所以，本专栏将从函数式编程角度来再看 JavaScript 精要，欢迎关注！传送门

前言

在 JS 中谈到 “响应式” ，你会想起什么？

1. 最初的 Object.observe ，已经被弃用了。。。

3. 还有 Object.defineProperty，它是 Vue2 响应式的核心。

2. 后来，ES6 有 Proxy 劫持了，很棒，Vue3 就是基于它的。

4. 再有，React 一词的中文就是“反应”、“响应”的意思，hooks 是 react 的最新“响应式”的解决方案；

还有吗？ —— 其实在原生 JS 中还有~

5. 比如 addEventListener，也是一种响应式吧，当目标元素被点击后，就会通知一个回调函数，进行特定的操作。

var handler = (e) => {
    console.log(e);
    document.body.removeEventListener('click', handler);
}

document.body.addEventListener('click', handler);

6. 还有，比如考察 Event loop ，常要背的微任务：MutationObserver 一定也别忘记。

// 得到要观察的元素
var elementToObserve = document.querySelector("#targetElementId");

// 创建一个叫 `observer` 的新 `MutationObserver` 实例，
// 并将回调函数传给它
var observer = new MutationObserver(function() {
    console.log('callback that runs when observer is triggered');
});

// 在 MutationObserver 实例上调用 `observe` 方法，
// 并将要观察的元素与选项传给此方法
observer.observe(elementToObserve, {subtree: true, childList: true});

7. 还有，设计模式中常问的“观察者模式”，这个面试常考。

class Producer {
    constructor() {
        this.listeners = [];
    }
    addListener(listener) {
        if(typeof listener === 'function') {
            this.listeners.push(listener)
        } else {
            throw new Error('listener 必须是 function')
        }
    }
    removeListener(listener) {
        this.listeners.splice(this.listeners.indexOf(listener), 1)
    }
    notify(message) {
        this.listeners.forEach(listener => {
            listener(message);
        })
    }
}

var egghead = new Producer(); 

function listener1(message) {
    console.log(message + 'from listener1');
}
function listener2(message) {
    console.log(message + 'from listener2');
}
egghead.addListener(listener1); // 注册监听
egghead.addListener(listener2);
egghead.notify('A new course!!') // 执行

// a new course!! from listener1
// a new course!! from listener2

代码可复制在控制台中调试。

通过回顾以上 7 点，“抛开其它不谈，这个响应式就没什么问题吗？” 不得不承认：响应式思想根植在前端 Script 和 DOM 的交互中
我们进一步想想：为什么是响应式？ 噢，其实，不为别的，就是为了偷懒！

偷懒的点在于，我们不想手动去触发函数的回调，设置响应式正是为了摆脱在时间上有异步操作而带来的困扰。

“我不管你什么时候操作，只要你操作了，就去触发XXX...”

响应式可以玩出各种各样的花来，这些其实就像是同一个事物在不同角度的展现。就像小学的那篇课文：《画杨桃》一样。关键在于你怎么看，是在其中的一面看，还是以全局视角来看。

按照这个思路继续往前，介绍今天的主角，基于响应式的新的花样：Observable，—— 它是 RxJS 的最最基础、最最核心的东西。

Observable 序列

整个 RxJS 最最基础的概念之一就是 Observable

什么是 Observable ？

网上看过很多解释，都不如人意，本瓜最后得出结论，不如就将其直接理解为一个序列。

什么是序列？

数组可能是我们用的最多的序列了。

你知道在 JS 中，数组还能这样迭代吗？

var arr = [1, 2, 3];
var iterator = arr[Symbol.iterator]();

iterator.next();
// { value: 1, done: false }
iterator.next();
// { value: 2, done: false }
iterator.next();
// { value: 3, done: false }
iterator.next();
// { value: undefined, done: true }

即使，不用 Symbol.iterator，我们也可以自己写一个迭代数组的方法。

自制 Iterator：

class IteratorFromArray {
    constructor(arr) {
        this._array = arr;
        this._cursor = 0;
    }
  
    next() {
        return this._cursor < this._array.length ?
        { value: this._array[this._cursor++], done: false } :
        { done: true };
    }
}

var iterator = new IteratorFromArray([1,2,3]);

iterator.next();

有一个 next 方法，返回 {value:val,done:false} 或者 {done:true}

这样看来，Iterator Pattern 似乎不难，但对比数组遍历它同时带来了两个优势：

它渐进式取值的特性可以拿来做延迟运算(Lazy evaluation)，让我们能用它来处理特殊结构（前面文章提过）；
因为 iterator 本身是序列，所以可以作所有阵列的运算方法像 map, filter... 等；

这个就厉害啦，这意味着 IteratorFromArray 函数还能再进一步处理：比如用 map 的思路：

class IteratorFromArray {
    constructor(arr) {
        this._array = arr;
        this._cursor = 0;
    }
  
    next() {
        return this._cursor < this._array.length ?
        { value: this._array[this._cursor++], done: false } :
        { done: true };
    }
    
    map(callback) {
        const iterator = new IteratorFromArray(this._array);
        return { 
            next: () => {
                const { done, value } = iterator.next();
                return {
                    done: done,
                    value: done ? undefined : callback(value)
                }
            }
        }
    }
}

var iterator = new IteratorFromArray([1,2,3]);
var newIterator = iterator.map(value => value + 3);

newIterator.next();
// { value: 4, done: false }
newIterator.next();
// { value: 5, done: false }
newIterator.next();
// { value: 6, done: false }

“不是讲 Observable 吗，怎么讲 Iterator 去了。。。”

—— Observable 和 Iterator 很像、很像

它们有一样的共性，即：它们都是渐进式取值，以及适用阵列的运算。

要说其唯一的区别可能是，Observable 序列更侧重于在“时间”这个维度上描述，即 Observable 的值会随着时间进行推送。

Observable 执行

以下所有介绍的 Observable 代码示例都可以在 jsfiddle 下运行

cdn 依赖是：https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.1/Rx.js

同步和异步

我们先测一个不带时间状态的同步的 Observable

在控制台依次输出：

测试地址

再测一个带时间状态的 Observable

同步结束后，执行异步的回调。

测试地址

细心的你一定发现了 subscribe 关键字的调用。subscribe 就是用来执行 Observable 的，就像是调用一个 function。

通常 subscribe 参数中的对象有三个值，分别是：next、error、complete，对应 observer 的三个状态：next、error、complete；

var observable = Rx.Observable
    .create(function (observer) {
            observer.next('Jerry');
            observer.next('Anna');
                        observer.complete();
    })
    
observable.subscribe({
    next: function(value) {
        console.log(value);
    },
    error: function(error) {
        console.log(error)
    },
    complete: function() {
        console.log('complete')
    }
})

测试地址

觉得理解起来麻烦，就通俗认为 subscribe 就是来处理 observer.next 的值的~

操作符

上述就是最简单的 Observable 推送值、取值的过程。

接下来，简单认识下如何新建 Observable 以及转换 Observable 。（都知道 RxJS 操作符很强大，它们其实大部分都是来操作 Observable 的。）

新建 Observable

Observable 有许多创建实例的方法，介绍最常见的几个~

create

create 前面都用的是这个，直接创建；

of

当我们想要同步的传递多个值时，可以用 of 这个 operator 来作简洁的表达

var source = Rx.Observable.of('Jerry', 'Anna');

source.subscribe(console.log);

测试地址

from

还可以用 from 来接收数组，创建 Observable

var arr = ['Jerry', 'Anna', 123, 456, 'juejin'] 
var source = Rx.Observable.from(arr);
source.subscribe(console.log);

测试地址

fromEvent

fromEvent 可以新建一个事件的 Observable

var source = Rx.Observable.fromEvent(document.body, 'click');

还有比如 fromEventPattern 可以新建类事件 Observable ，比如同时具有添加监听、移除监听的方法。

interval

每隔一定时间间隔产生值的 Observable

var source = Rx.Observable.interval(1000);

转换 Observable

常见的转换 Observable 比如像是 map, filter, contactAll......等等，所有这些函数都会拿到原本的observable 并回传一个新的observable。

map

// 生成一个间隔为1秒的时间序列，每秒输出的值为秒数*2

var source = Rx.Observable.interval(1000);
var newest = source.map(x => x*2); 
newest.subscribe(console.log);

// 0
// 2
// 4
// 6
...

测试地址

filter

// 生成一个间隔为1秒的时间序列，过滤掉奇数秒

var source = Rx.Observable.interval(1000);
var newest = source.filter(x => x % 2 === 0); 
newest.subscribe(console.log);

// 0
// 2
// 4
// 6
..

测试地址

concatAll

有时我们的 Observable 送出的元素又是一个 observable，就像是二维阵列，阵列里面的元素是阵列。

这时我们就可以用 concatAll 把它摊平成一维阵列，concatAll 把所有元素 concat 起来。

// 生成一个间隔为1秒的时间序列，取前 5 个值，
// 再生成一个间隔为 0.5 秒的时间序列，取前 2 个值
// 再生成一个间隔为 2 秒的时间序列，取前 1 个值
// 把这些值返回给一个 Observable，相当于是二维的 Observable，再用 concatAll 拉平；

var obs1 = Rx.Observable.interval(1000).take(5);
var obs2 = Rx.Observable.interval(500).take(2);
var obs3 = Rx.Observable.interval(2000).take(1);
var source = Rx.Observable.of(obs1, obs2, obs3);
var example = source.concatAll();
example.subscribe(console.log);
// 0
// 1
// 2
// 3
// 4
// 0
// 1
// 0

时间线的弹珠图示意：（ps: 不懂弹珠图的可看下一小节释义）

source : (o1                 o2      o3)|
           \                  \       \
            --0--1--2--3--4|   -0-1|   ----0|
            
                    concatAll()
                    
example: --0--1--2--3--4-0-1----0|

测试地址

observable 操作的 API 有很多，一下子就记全、记清也是不现实的，我们应该 在学中用，在用中记，多看几遍就熟了，常用、关键的方法其实也不多。 rx.js.org-操作符分类

弹珠图

我们在传达事物时，文字其实是最糟的手段，虽然文字是我们平时沟通的基础，但常常千言万语也比不过一张清楚的图。

我们把描绘 observable 的图示称为弹珠图。

用 - 来表达一小段时间，这些 - 串起就代表一个observable。|则代表observable 结束

比如：

var source = Rx.Observable.interval(1000);

弹珠图：

-----0-----1-----2-----3--...

var source = Rx.Observable.interval(1000);
var newest = source.map(x => x + 1);

弹珠图：

source: -----0-----1-----2-----3--...
            map(x => x + 1)
newest: -----1-----2-----3-----4--...

最常用操作

当操作比较复杂的时候，需要用到弹珠图来理解，https://rxviz.com/ 这个网站可以专门来绘制弹珠图。

merge

merge 用来合并 observable

var source = Rx.Observable.interval(500).take(3);
var source2 = Rx.Observable.interval(300).take(6);
var example = source.merge(source2);
example.subscribe(console.log);

source : ----0----1----2|
source2: --0--1--2--3--4--5|
            merge()
example: --0-01--21-3--(24)--5|

测试地址

可以看到 merge 和 concatAll 有区别：concatAll 是一个 Observable 彻底走完，再走下一个，merge 是同时跑，不管谁先推送值，都将其先取。

combineLatest

它会取得各个 observable 最后送出的值，再输出成一个值；

var source = Rx.Observable.interval(500).take(3);
var newest = Rx.Observable.interval(300).take(6);
var example = source.combineLatest(newest, (x, y) => x + y);
example.subscribe(console.log);

source : ----0----1----2|
newest : --0--1--2--3--4--5|
    combineLatest(newest, (x, y) => x + y);
example: ----01--23-4--(56)--7|

测试地址

withLatestFrom

withLatestFrom 运作方式跟 combineLatest 有点像，只是他有主从的关系，只有在主要的 observable 送出新的值时，才会执行 callback；

var main = Rx.Observable.from('hello').zip(Rx.Observable.interval(500), (x, y) => x);
var some = Rx.Observable.from([0,1,0,0,0,1]).zip(Rx.Observable.interval(300), (x, y) => x);
var example = main.withLatestFrom(some, (x, y) => {
    return y === 1 ? x.toUpperCase() : x;
});

example.subscribe(console.log);

main   : ----h----e----l----l----o|
some   : --0--1--0--0--0--1|
withLatestFrom(some, (x, y) =>  y === 1 ? x.toUpperCase() : x);
example: ----h----e----l----L----O|

测试地址

实战

OK，理论讲太多，也会乏味。就上面的 api 其实就已经够了，我们可以通过他们用短短几行代码实现复杂的功能。

基础拖拉

短短 15 行代码就可以实现一个基础的拖拽功能。在线测试地址

const dragDOM = document.getElementById('drag');
const body = document.body;

const mouseDown = Rx.Observable.fromEvent(dragDOM, 'mousedown');
const mouseUp = Rx.Observable.fromEvent(body, 'mouseup');
const mouseMove = Rx.Observable.fromEvent(body, 'mousemove');

mouseDown
  .map(event => mouseMove.takeUntil(mouseUp))
  .concatAll()
  .map(event => ({ x: event.clientX, y: event.clientY }))
  .subscribe(pos => {
    dragDOM.style.left = pos.x + 'px';
    dragDOM.style.top = pos.y + 'px';
  })

思路：

获取 dragDOM
用 fromEvent 创建 mousedown、mouseup、mousemove 事件。
当第一次 mouseDown 时，监听 mouseMove，直到 mouseUp；
这个过程中，修改 dragDOM 的left、top 值；

只要能看懂 Observable operators，代码可读性非常高。既简洁，又易维护。

视频拖拉

接着拖拽的需求，再进一步。

我们在网页中看视频的时候，经常遇到这样的场景：下拉滚动条，视频缩放到右小角，并且可以拖拽。

用 RxJS Observable，35 行代码即能实现：

const video = document.getElementById('video');
const anchor = document.getElementById('anchor');

const scroll = Rx.Observable.fromEvent(document, 'scroll');
const mouseDown = Rx.Observable.fromEvent(video, 'mousedown')
const mouseUp = Rx.Observable.fromEvent(document, 'mouseup')
const mouseMove = Rx.Observable.fromEvent(document, 'mousemove')

const validValue = (value, max, min) => {
    return Math.min(Math.max(value, min), max)
}

scroll
.map(e => anchor.getBoundingClientRect().bottom < 0)
.subscribe(bool => {
    if(bool) {
        video.classList.add('video-fixed');
    } else {
        video.classList.remove('video-fixed');
    }
})

mouseDown
    .filter(e => video.classList.contains('video-fixed'))
    .map(e => mouseMove.takeUntil(mouseUp))
    .concatAll()
    .withLatestFrom(mouseDown, (move, down) => {
        return {
            x: validValue(move.clientX - down.offsetX, window.innerWidth - 320, 0),
            y: validValue(move.clientY - down.offsetY, window.innerHeight - 180, 0)
        }
    })
    .subscribe(pos => {
        video.style.top = pos.y + 'px';
        video.style.left = pos.x + 'px';
    })

思路分为 3 部分：