三连弹！原生实现异步处理利器 —— Observable

掘金安东尼

发布于 2022-09-19 10:59:41

2940

发布于 2022-09-19 10:59:41

文章被收录于专栏：掘金安东尼

这是我参与11月更文挑战的第18天，活动详情查看：2021最后一次更文挑战

本篇译自：https://javascript.plainenglish.io/how-to-use-observables-with-vanilla-javascript

接上两篇：

我们认识了 ES7 处理异步的一大利器 —— Observable

本篇带来用原生实现 Observable，一探内部究竟！！

实现步骤解析如下：

Observable 应该是一个 Subject 新对象类型；
并且 Subject 对象有subscribe和next两个函数；
subscribe 由 observers （观察者）调用，来订阅可观察的数据流；
next由 Subject owner 调用，实现推送/发布新数据；
Subject owner 应该知道 observers 何时监听数据生效；
另外，Subject owner 也应该知道 observers 何时不再监听了；
再看 observer，它也应该可以随时被取消；
可以定义一个新的对象，Subscription；
Subscription 包含 unsubscribe 函数；
每个 observer 想要停止监听来自 Subject 的数据流时，可以调用unsubscribe实现；

有了以上思路之后，我们可以进一步来看代码实现~

代码实现：

Subscription

let Subscription = function(handlerId, unsubscribeNotificationCallback) {
    let self = this;

    self.unsubscribe = () => {
        if(unsubscribeNotificationCallback) {
            unsubscribeNotificationCallback(handlerId);
        }
    };
    
    return self;
};

注意：Subscription 仅在 unsubscribe 被调用时，通知 Subject；

再看 Subject 完整代码实现：

let Subject = function(subscribersStateChangeNotificationCallback) {
    let self = this;
    
    let handlers = {};
    
    Object.defineProperty(self, "subscribersFound", {
        get() {
            let found = false;
            
            for(const prop in handlers) {
                if(handlers.hasOwnProperty(prop)) {
                    found = true;
                    break;
                }
            }
            
            return found;
        }
    });
    
    Object.defineProperty(self, "subscribersCount", {
        get() {
            let count = 0;
            
            for(const prop in handlers) {
                if(handlers.hasOwnProperty(prop)) {
                    count++;
                }
            }
            
            return count;
        }
    });
    
    let unsubscribeNotificationCallback = (handlerId) => {
        if(handlerId &amp;&amp; handlerId !== '' &amp;&amp; handlers.hasOwnProperty(handlerId)) {
            delete handlers[handlerId];
            
            if(subscribersStateChangeNotificationCallback &amp;&amp; !self.subscribersFound) {
                subscribersStateChangeNotificationCallback(false);
            }
        }
    };
    
    self.subscribe = (handler) => {
        let handlerId = createGuid();
        handlers[handlerId] = handler;
        
        if(subscribersStateChangeNotificationCallback &amp;&amp; self.subscribersCount === 1) {
            subscribersStateChangeNotificationCallback(true);
        }
        
        return new Subscription(handlerId, unsubscribeNotificationCallback);
    };
    
    self.next = (data) => {
        for(const handlerId in handlers) {
            handlers[handlerId](data);
        }
    };
    
    return self;
};

let createGuid = function() {  
   return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function(c) {  
      var r = Math.random()*16|0, v = c === 'x' ? r : (r&amp;0x3|0x8);  
      return v.toString(16);  
   });  
};

Subject Owner 实现：

.
.
.
let subscribersStateChangeNotificationCallback = (subscriberFound) => {
    if(!subscriberFound &amp;&amp; isNowWatching) {
        stopWatching();
        isNowWatching = false;
    } else if(subscriberFound &amp;&amp; !isNowWatching) {
        startWatching();
    }
};

self.data = new Subject(subscribersStateChangeNotificationCallback);
.
.
.
self.data.next(self.snapshot.data);
.
.
.

Observer 实现：

.
.
.
const dashboardServiceSubscription = myDashboardService.data.subscribe((data) => {
    ...
});
.
.
.
dashboardServiceSubscription.unsubscribe();
.
.
.

小结：我们可以看到实现关键是 Subject 对象，更重要的是发布和订阅的过程！当然，也不能忘了 取消订阅 的功能；

发布和订阅模式来处理异步可以忽视掉时间这个维度，就是不用管时间上的先后，就保证了顺序！这一点，在前面一篇函数式编程中也讲过：《XDM，JS如何函数式编程？看这就够了！（六）》 —— 减少时间状态！

不得不说，都是相通的~~ Σ(⊙▽⊙"a

OK，就是这样！这就是用原生模拟 Observable 的实现过程！

撰文不易，点赞鼓励👍👍👍👍👍👍