Angular 个人深究（二）【发布与订阅】

Angular 个人深究（二）【发布与订阅】

---

1. 再入正题之前，首先说明下[ 发布与订阅模式]（也叫观察者模式）

　　1) 定义：定义了一种一对多的依赖关系，让多个观察者对象同时监听某一个主题对象。这个主题对象在状态发生变化时会通知所有观察者对象，使它们能够自动更新。

　　2) 结构图：

　　3) 解释：Subject类，可翻译为主题或抽象的通知者，一般是抽象类或者接口。Observer是抽象观察者，为所有的具体观察者定义一个接口。ConcreteSubject类，叫做具体通知者。ConcreteObserver是具体的观察者。

　　5) 代码实现：（用TypeScript实现）

/* 代码使用Typescript 编写，Angular中使用的他，需要编译成Javascript代码然后使用node js 运行*/

//抽象观察者
abstract class abstract_observer{
    name:string;
    message:string;
    constructor(name:string){
        this.name = name;
    }
    abstract handle_publish(msg);
}
//抽象主题
abstract class abstract_subject{
    subject_name:string;
    observers: abstract_observer[]= [];
    current_message:string;
    constructor(name:string){
        this.subject_name = name;
    }
    //提供订阅入口
    subscribe(observer:abstract_observer){
        this.observers.push(observer);
        console.log(observer.name+"成功订阅"+this.subject_name);
    }
    //提供取消订阅入口
    unsubscribe(observer:abstract_observer ){
        this.observers.splice(this.observers.indexOf(observer),1);
        console.log(observer.name+"成功取消订阅"+this.subject_name);
    }
    update_message(msg){
        this.current_message = msg;
    }
    //发布消息
    publish(){
        console.log("Server 开始发布消息");
        for (let observer of this.observers){
            console.log("发布消息给"+observer.name+"!");
            observer.handle_publish(this.current_message);
        }
        console.log("所有订阅"+this.subject_name+"的人已经收到消息！");
    }
}
//具体观察者
class concrete_observer extends abstract_observer{
    constructor(name:string){
        super(name);
    }
    handle_publish(msg){
        this.message = msg;
        console.log(this.name+": 已经接到消息:"+this.message);
    }
}
//具体股票主题
class concrete_subject_gupiao extends abstract_subject{
    publish(){
        console.log('发送股票新消息');
        super.publish();
    }
    update_message(updatemsg){
        console.log("股票消息更新:"+updatemsg);
        super.update_message(updatemsg);
    }
}
//具体NBA主题
class concrete_subject_nba extends abstract_subject{
    publish(){
        console.log('发送NBA新消息');
        super.publish();
    }
    update_message(updatemsg){
        console.log("NBA消息更新："+updatemsg);
        super.update_message(updatemsg);
    }
}
//订阅发布主逻辑
var observer1 = new concrete_observer("小明");
var observer2 = new concrete_observer("小强");
var observer3 = new concrete_observer("小红");
var subject_gupiao = new concrete_subject_gupiao("股票");
var subject_nba = new concrete_subject_nba("NBA");
console.log("小明订阅了股票");
subject_gupiao.subscribe(observer1);
console.log("小明订阅了NBA");
subject_nba.subscribe(observer1);
console.log("小强订阅了股票");
subject_gupiao.subscribe(observer2);
console.log("小红订阅了NBA");
subject_nba.subscribe(observer3);
console.log("---------------------------------");
subject_gupiao.update_message("大盘下跌");
subject_nba.update_message("骑士总冠军");
console.log("---------------------------------");
subject_gupiao.publish();
console.log("---------------------------------");
subject_nba.publish();
console.log("---------------------------------");
console.log("小明取消订阅股票");
subject_gupiao.unsubscribe(observer1);
console.log("小明取消订阅NBA");
subject_nba.unsubscribe(observer1);
console.log("---------------------------------");
subject_gupiao.update_message("大盘上涨");
subject_nba.update_message("骑士蝉联总冠军");
console.log("---------------------------------");
subject_gupiao.publish();
console.log("---------------------------------");
subject_nba.publish();
console.log("---------------------------------");

　　　　输入结果如下：

小明订阅了股票
小明成功订阅股票
小明订阅了NBA
小明成功订阅NBA
小强订阅了股票
小强成功订阅股票
小红订阅了NBA
小红成功订阅NBA
---------------------------------
股票消息更新:大盘下跌
NBA消息更新：骑士总冠军
---------------------------------
发送股票新消息
Server 开始发布消息
发布消息给小明!
小明: 已经接到消息:大盘下跌
发布消息给小强!
小强: 已经接到消息:大盘下跌
所有订阅股票的人已经收到消息！
---------------------------------
发送NBA新消息
Server 开始发布消息
发布消息给小明!
小明: 已经接到消息:骑士总冠军
发布消息给小红!
小红: 已经接到消息:骑士总冠军
所有订阅NBA的人已经收到消息！
---------------------------------
小明取消订阅股票
小明成功取消订阅股票
小明取消订阅NBA
小明成功取消订阅NBA
---------------------------------
股票消息更新:大盘上涨
NBA消息更新：骑士蝉联总冠军
---------------------------------
发送股票新消息
Server 开始发布消息
发布消息给小强!
小强: 已经接到消息:大盘上涨
所有订阅股票的人已经收到消息！
---------------------------------
发送NBA新消息
Server 开始发布消息
发布消息给小红!
小红: 已经接到消息:骑士蝉联总冠军
所有订阅NBA的人已经收到消息！
---------------------------------

　　　　说明：实现了订阅与发布不同主题的情况，并可以取消订阅。

　　4) 分享一个设计模式很好的书籍【大话设计模式——程杰】

　　　　百度网盘下载地址: https://pan.baidu.com/s/1iyUe5p0yAHNq3Hw2Z4L1Qg 提取码：n1pd

　　　　感兴趣的可以去看看。

2.下面开始说明Angular中的发布与订阅

　　注：本人在使用angular时，为了实现父子组件之间的变量传递，才研究到了其中的发布与订阅

• 功能实现：子组件更改数据到父组件（方法一）

//test1.component.ts
import { Component, OnInit } from '@angular/core';
import {EmitserviceService} from '../emitservice.service';
@Component({
  selector: 'app-test1',
  templateUrl: './test1.component.html',
  styleUrls: ['./test1.component.css']
})
export class Test1Component implements OnInit {
    data_from_child:string;
  constructor(private emitter:EmitserviceService) {}
  ngOnInit() {
        this.emitter.emitter.subscribe((data)=>{
            console.log(data);
            this.data_from_child = data;
        });
  }
}
//test2.component.ts
import { Component, OnInit,EventEmitter } from '@angular/core';
import {EmitserviceService} from '../emitservice.service';
@Component({
  selector: 'app-test2',
  templateUrl: './test2.component.html',
  styleUrls: ['./test2.component.css']
})
export class Test2Component implements OnInit {
  constructor(private emitter:EmitserviceService) {}
  ngOnInit() {
  }
    onTest(value:string){
        this.emitter.emitter.emit(value);
    }
}
//emitservice.service.ts
import { Injectable,EventEmitter } from '@angular/core';

@Injectable({
  providedIn: 'root'
})
export class EmitserviceService {
    public emitter:any;
    constructor() { 
        this.emitter = new EventEmitter ; 
    }
}
//test1component.html
<div>
　<div><h1>这是父组件test1</h1></div>
    <h2>子组件的数据(子到父)</h2>
    <h3>显示子组件的数据：{{data_from_child}}</h3>
　<div>
    <app-test2></app-test2> 
　</div>
</div>
//test2.component.html
<div><h1>这是子组件test2</h1></div>
输入数据传递到父组件：
<input type="text" (change)="onTest($event.target.value)">

• 底层代码探究：

　　　　需要用到的文件有：

　　 ode_modules@angularcorefesm2015core.js

　　 ode_modules xjs\_esm5internalSubject.js

　　 ode_modules xjs\_esm5internalSubscriber.js

　　 ode_modules xjs\_esm5internalObservable.js

　　 ode_modules xjs\_esm5internalutil oSubscriber.js

---

core.js

//core.js 部分代码
//实际上是继承了Subject类，是Angular的部分 Subject是Rxjs的部分
class EventEmitter extends Subject {
//定义了发射数据函数，并调用Subject的next方法
    emit(value) { super.next(value); }
//定义了 订阅 方法，并调用了Subject 依赖的Observable 的 subscibe方法
    subscribe(generatorOrNext, error, complete) {
    //省略部分代码
        const /** @type {?} */ sink = super.subscribe(schedulerFn, errorFn, completeFn);
        //省略部分代码
    }
}

　　　　core.js代码 是angular的代码，实际上就是定义了class EvemtEmitter 这个类，方便angluar中的使用，内部提供了emit与subscribe方法实现发布与订阅功能，实际上也是使用了rxjs的subject类

Subject.js

var Subject = /*@__PURE__*/ (function (_super) {
    // Subject 继承 Observable
    tslib_1.__extends(Subject, _super);
        console.log(_super);
    function Subject() {
        var _this = _super.call(this) || this;
        _this.observers = [];
        _this.closed = false;
        _this.isStopped = false;
        _this.hasError = false;
        _this.thrownError = null;
        return _this;
    }
    Subject.prototype[rxSubscriberSymbol] = function () {
        return new SubjectSubscriber(this);
    };
    Subject.prototype.lift = function (operator) {
        var subject = new AnonymousSubject(this, this);
        subject.operator = operator;
        return subject;
    };
    //next 函数，将发布消息给observer
    Subject.prototype.next = function (value) {
        if (this.closed) {
            throw new ObjectUnsubscribedError();
        }
        if (!this.isStopped) {
            var observers = this.observers;
            var len = observers.length;
            var copy = observers.slice();
            for (var i = 0; i < len; i++) {
                copy[i].next(value);
            }
        }
    };
//错误处理
    Subject.prototype.error = function (err) {
        if (this.closed) {
            throw new ObjectUnsubscribedError();
        }
        this.hasError = true;
        this.thrownError = err;
        this.isStopped = true;
        var observers = this.observers;
        var len = observers.length;
        var copy = observers.slice();
        for (var i = 0; i < len; i++) {
            copy[i].error(err);
        }
        this.observers.length = 0;
    };
//手动完成
    Subject.prototype.complete = function () {
        if (this.closed) {
            throw new ObjectUnsubscribedError();
        }
        this.isStopped = true;
        var observers = this.observers;
        var len = observers.length;
        var copy = observers.slice();
        for (var i = 0; i < len; i++) {
            copy[i].complete();
        }
        this.observers.length = 0;
    };
//取消订阅
    Subject.prototype.unsubscribe = function () {
        this.isStopped = true;
        this.closed = true;
        this.observers = null;
    };
    Subject.prototype._trySubscribe = function (subscriber) {
        if (this.closed) {
            throw new ObjectUnsubscribedError();
        }
        else {
            return _super.prototype._trySubscribe.call(this, subscriber);
        }
    };
        //在observable中调用_subscribe 将observer添加到this.Observable中
    Subject.prototype._subscribe = function (subscriber) {
            console.log(subscriber);
        if (this.closed) {
            throw new ObjectUnsubscribedError();
        }
        else if (this.hasError) {
            subscriber.error(this.thrownError);
            return Subscription.EMPTY;
        }
        else if (this.isStopped) {
            subscriber.complete();
            return Subscription.EMPTY;
        }
        else {
            this.observers.push(subscriber);
            return new SubjectSubscription(this, subscriber);
        }
    };
    Subject.prototype.asObservable = function () {
        var observable = new Observable();
        observable.source = this;
        return observable;
    };
    Subject.create = function (destination, source) {
        return new AnonymousSubject(destination, source);
    };
    return Subject;
}(Observable));
export { Subject };

　　　　Subject.js中的subject类定义了 next、complete、error、_subscirbe等方法，实现订阅与发布

Subscriber.js：在Subject的Observers中存放的就是Subscriber 理解为订阅者。

Observable.js：提供了订阅方法。

toSubscriber.js：将返回Subscriber对象

• 整体的代码流程

1.test1组件中使用core.js的方法订阅了数据，并传入一个处理订阅数据的函数。

2.在core.js中订阅函数会调用Observable的订阅函数Subscribe，在这个函数中将传入的处理函数转成 subscirber

3.由于这种方式Observable的operator是空的，所有会回调到subject.js中的_subscribe()方法将subscriber放到subject的observers中

4.test2组件中某个定义是事件触发core.js中的emit方法

5.emit方法又调用了subject的next方法

6.调用subscriber中的next方法

7.调用subscirber中的_next方法

8.调用SafeSubscriber中的next方法（在subscriber的构造函数中会将destination转化成SafeSubscriber）

9.调用SafeSubscriber中的_tryOrUnsub方法,执行test1传进来的函数

• 总结：

关于RXJS的Subject还有很多扩展的地方，本文意在完成父子组件之间的数据传递。如有任何概念性错误望指正，谢谢！