我只用RxJS，却搞定了三大框架的跨组件通信，甚至还能适用于Java（三、React篇）
开篇
本文代码git地址：Crimson/rxjs-cross-component-communication (gitee.com)
承接上回，我用RxJS完成了在Angular与Vue中的跨组件通信，本文就来讲讲在React中，怎么将这个东西移植过去。
简单重述下本系列所讲述的背景：
我作为六年前从Java入行的程序员，做过后端，做过原生安卓，做过Java桌面，后来前后端分离流行了之后开始接触Angular直到现在。从RxJava用到RxJS，对这东西可以说是天天打交道，再熟悉不过了。
不过在这里我们不谈底层，不谈源码，也不谈RxJS一共有哪些API。我们只谈如何用RxJS去实现一个适用于前端三个框架的跨组件通信的功能。移动端开发常说，一份代码，全平台运行。在这里我们也试试看，一份代码，全框架通用。
与上篇Vue中讲到的相同，对于React初学者，我希望同学们要好好学习这篇文章中的思想，怎么用面向对象，怎么用最基础的设计模式，以及我所认为的发布订阅，我是怎么设计它的模型，等等。同时，在我看来，函数式不是万能的，前端开发者也必须要掌握面向对象编程。作为从Java出身的前端程序员，在6年开发生涯中，面向对象思维在前端帮我解决了不少复杂业务。
说个题外话，不少人认为TypeScript难学，不过在我看来它有什么难学的呢？无非就是你从前端入门，直接学习TS，在此之前没接触过强类型语言，也不了解面向对象的编程思维，当然就难学了。我不建议你在没有掌握其他语言的情况下，去走大前端路线。不要求你去掌握C++，有这个精力去学学Java，学学C#，你会发现TS真的不难，前端技术变来变去也就那样。把自己的技术栈限制在浏览器范围内，是永远也接触不到计算机原理与软件工程思维的。
为什么选择RxJS
本文就不去介绍RxJS了。有不了解的同学可以看看上篇文字，有简单介绍RxJS。Vue篇中介绍RxJS的传送门
至于为什么选择RxJS，因为一开始它就是Angular中自带的一个插件，并且我的跨组件通信也是基于RxJS来实现的。不得不说RxJS做异步流是真的挺方便。并且RxJS仅仅只是一个单纯的插件，站在它的角度来看并没有和框架有过深入的结合。简单来说，RxJS属于原生JS的范畴，你把它放哪都能玩得起来。
在几年前最初造这个轮子的时候，当时的代码和今天的差距很大。今天大家接下来读到的代码是经过我多次重构的结果。
本文的目的，并不在于推翻Redux，或者是Pinia，再或者是给出一个新技术，让大家去卷去学习。作者的目的仅仅只是一个简简单单的技术分享，以及多年开发总结的心得体会。一次简单的探索，一次大胆的尝试。作者更多的是希望各位同学们，不要用了框架，就忘了原生，不要学了函数式，就丢弃面向对象。
我对发布订阅的理解
这里再次讲讲我是怎么理解发布订阅的，方便第一次点进来的小伙伴们能够快速了解本文所讲述的内容。
说到发布订阅，我第一个想到的就是消息队列。我刚入行的时候，公司是做物联网项目的，大致业务就是设备通过消息队列上报位置到服务端，然后服务端通过websocket发送给前端页面在地图上显示。好巧不巧，我第二份工作依然是做这东西，业务也极其相似。于是就在最初的开发生涯中结识了两个消息队列：RabbitMQ与MQTT。他们有一个共同点：基于Topic的发布订阅模式。所以受到这个启发，我在第一次用Angular做项目的时候，用RxJS简单实现了该模式的功能。后来经过多个项目的迭代优化，于是就有了今天这篇文章，分享出来。
我在Vue篇中也单独把这一段提出来讲了讲。Vue篇中讲解发布订阅的传送门
结构设计
以基于Topic的发布订阅模式为例，我们照着它的功能去简单的实现（不一定与它完全相同，只参考大致的结构）：
- 它有一个服务端
- 有很多设备去连接这个服务端
- 每个设备会订阅一个或多个Topic
- 给服务端根据Topic发送消息，服务端会将消息发送给订阅该Topic的设备
接下来使用简单的设计模式在代码中实现它（这也是为什么我要去使用面向对象）：
- 服务端只有一个，它是单例的
- 每个设备看作为客户端，客户端有多个，它是多例的
- 使用依赖注入，让框架去管理服务端，这样你就获得了一个可以在各处调用的服务端单例了
- 把页面组件比作设备，相当于每个组件都是一个客户端，各自根据Topic去订阅消息
代码实现
废话不多说，直接在代码中展现出来吧！
简单搭建一个React项目，安装好RxJS，并准备好各个组件。创建过程就不赘述了。安装RxJS也很简单：
npm install rxjs
简单地搭建下，为了简略配置，专注主要功能，本篇就不使用scss了。与上回保持一致，我们直接看代码目录和页面组件结构：
简单描述下：
- 在创建好的React项目中，创建两个文件夹，一个是rx，一个是view
- rx目录存放通过RxJS实现的发布订阅相关代码
- view目录中为页面组件，也就是上图页面中所展示出来的各个组件的结构
基于此页面组件结构，接下来就来实现test-a组件与test-b组件之间的通信。
封装RxJS
本章节就来讲讲如何用RxJS封装一个基于Topic的发布订阅模式的工具。在这里我给它取名为inner-mq。其实我们直接把上回在Angular中写好的代码移植过来即可，相比于Vue，在React中使用也不需要修改import，只需要删掉@Injectable()注解。在本章我会重新讲讲设计方法。
服务端设计
服务端的作用就是对客户端进行管理，每个组件能够注册客户端，销毁客户端。以及将发布者发送的消息根据Topic转发给指定客户端。
除此之外我做了个简单的持久化，它的作用就是：
- 如果发布发生在订阅之前，可以暂时存储消息，等待有订阅后再将消息发出
- 可以实现某个消息，在每一次订阅的时候都去发送一次
这样一来，很好地解决了订阅和发布之间时机错过的问题，你就不用担心发布发生在订阅之前了。
rx/service/inner-mq.service.ts
export class InnerMqService {
private random = new Random(); // 使用种子随机数生成唯一ID
private clients = new Map<string, InnerMqClient>(); // 客户端
private topic2clients = new Map<Topic, Map<string, InnerMqClient>>(); // 根据Topic存储的客户端
private persistentQueue = new Map<any, Array<{ type: PersistentType, data: any }>>(); // 持久化队列
constructor() {
}
/* 创建连接 */
public createClient(id?: string): InnerMqClient {
if (id == null) {
id = Random.generateCharMixed(20) + '_' + this.random.nextInt(2147483647);
}
if (this.clients.has(id)) {
throw 'Client ID重复';
}
let client = new NormalInnerMqClient(id, {
onSubscribe: (topic, subject) => {
// 根据topic存储client
if (this.topic2clients.get(topic) == null) {
this.topic2clients.set(topic, new Map<string, InnerMqClient>);
}
this.topic2clients.get(topic)?.set(client.getId(), client);
// 完成存储后执行其它回调方法
this.clientSubscribeCallback(client, topic, subject);
}
});
this.clients.set(id, client);
return client;
}
/* 销毁连接 */
public destroyClient(client: InnerMqClient): void {
// 删除客户端
this.clients.delete(client.getId());
// 删除根据Topic存储的客户端
for (let topic of this.topic2clients.keys()) {
this.topic2clients.get(topic)?.delete(client.getId());
}
client.destroy();
}
/* 发布 */
public pub(topic: Topic, msg: any, option?: { persistent: boolean, type: PersistentType }): void {
let published = false;
let clients = this.topic2clients.get(topic);
if (clients == null) {
published = false;
} else {
for (let client of clients.values()) {
if (client.isDestroyed()) {
published = false;
}
let subject = client.getSubject(topic);
if (subject != null && !subject.closed) {
subject.next(msg);
published = true;
} else {
published = false;
}
}
}
// 消息未发送，进行持久化存储
if (!published && (option && option.persistent)) {
if (this.persistentQueue.get(topic) == null) {
this.persistentQueue.set(topic, []);
}
this.persistentQueue.get(topic)?.push({ type: option.type, data: msg });
}
}
/* 客户端订阅回调 */
private clientSubscribeCallback(client: InnerMqClient, topic: Topic, subject: Subject<any>): void {
// 处理持久化消息
this.processPersistentQueue(topic, subject);
}
/* 处理持久化消息 */
private processPersistentQueue(topic: Topic, subject: Subject<any>): void {
let queue = this.persistentQueue.get(topic);
if (queue == null) {
return;
}
// 异步发送已持久化的消息
new Observable<boolean>((observer) => {
Promise.resolve().then(() => {
observer.next(true);
})
}).subscribe(() => {
if (queue == null) {
return;
}
for (let i = 0; i < queue.length; i++) {
switch (queue[i].type) {
case PersistentType.ON_ONCE_SUB:
subject.next(queue[i].data);
queue.splice(i, 1); // 将使后面的元素依次前移，数组长度减1
i--; // 如果不减，将漏掉一个元素
break;
case PersistentType.ON_EVERY_CLIENT_EVERY_SUB:
subject.next(queue[i].data);
break;
default:
break;
}
}
if (queue.length == 0) {
this.persistentQueue.delete(topic);
}
})
}
}
export enum PersistentType {
ON_ONCE_SUB, // 只进行一次缓存，一次sub后即删除
ON_EVERY_CLIENT_EVERY_SUB, // 持久化，对每个客户端的每一次该TOPIC的sub都发送信息
}
另外简单写一个随机数生成类，用于为每个客户端生成唯一ID：
rx/util/random.ts
/**
* 像Math.seededRandom这种伪随机数生成器叫做线性同余生成器（LCG, Linear Congruential Generator)，几乎所有的运行库提供的rand都是采用的LCG，形如：
* I n+1=aI n+c(mod m)
* 生成的伪随机数序列最大周期m，范围在0到m-1之间。要达到这个最大周期，必须满足：
* 1.c与m互质
* 2.a - 1可以被m的所有质因数整除
* 3.如果m是4的倍数，a - 1也必须是4的倍数
* 以上三条被称为Hull-Dobell定理。作为一个伪随机数生成器，周期不够大是不好意思混的，所以这是要求之一。因此才有了：a=9301, c = 49297, m = 233280这组参数，以上三条全部满足。
* */
export class Random {
private seed: number;
// 实例化一个随机数生成器，seed=随机数种子，默认当前时间
constructor(seed?: number) {
this.seed = (seed || Date.now()) % 999999999;
}
// 返回0~1之间的数
public next() {
this.seed = (this.seed * 9301 + 49297) % 233280;
return this.seed / 233280.0;
}
// 返回0~max之间的数
public nextInt(max: number) {
return Math.floor(this.next() * max);
}
// 静态方法：生成n位数字字母混合字符串
public static generateCharMixed(n: number) {
let chars = [
'1', '2', '3', '4', '5', '6', '7', '8', '9', '0', '-', '=',
'~', '!', '@', '#', '$', '%', '^', '&', '*', '(', ')', '+', '*',
'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm',
'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z',
'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M',
'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z',
];
let res = '';
for (let i = 0; i < n; i++) {
let id = Math.floor(Math.random() * chars.length);
res += chars[id];
}
return res;
}
}
客户端设计
客户端的作用很单一，就是单纯地订阅消息，即通过Topic接收发布者给服务端发过来的消息。
先设计一个接口，有了接口，就可以通过不同的实现类，去做不同类型的客户端。interface与implements是Java中很常用的一组关键字。
每个客户端拥有自己的唯一ID，由服务端生成并作为参数传入。
rx/client/inner-mq.client.ts
export interface InnerMqClient {
getId(): string;
getSubject(topic: Topic): Subject<any> | undefined;
isDestroyed(): boolean;
sub<T>(topic: Topic, subscribe: (e: T) => void): string;
stopSubByTopic(topic: Topic): void;
stopSubBySubscriptionId(id: string): void;
destroy(): void;
}
接着就是通过一个类去实现该接口，实现一个具体的客户端。
rx/client/impl/normal-inner-mq.client.ts
export class NormalInnerMqClient implements InnerMqClient {
private subjects: Map<Topic, Subject<any>> = new Map<Topic, Subject<any>>(); // 实例
private subscriptions: Array<{ id: string, topic: Topic, subscription: Subscription }> = []; // 订阅列表
private destroyed = false;
constructor(
private readonly id: string,
private callback: {
onSubscribe: (topic: Topic, subject: Subject<any>) => void
}
) {
}
public getId(): string {
return this.id;
}
public getSubject(topic: Topic): Subject<any> | undefined {
return this.subjects.get(topic);
}
public isDestroyed(): boolean {
return this.destroyed;
}
/* 订阅 */
public sub<T>(topic: Topic, subscribe: (e: T) => void): string {
let subject = this.subjects.get(topic);
if (subject == null) {
subject = new Subject<any>();
this.subjects.set(topic, subject);
}
let id = this.getSubscriptionId();
let subscription = subject.subscribe(res => subscribe(res));
this.subscriptions.push({ id: id, topic: topic, subscription: subscription });
this.callback.onSubscribe(topic, subject);
return id;
}
/* 取消订阅 By Topic */
public stopSubByTopic(topic: Topic): void {
for (let i = 0; i < this.subscriptions.length; i++) {
if (topic == this.subscriptions[i].topic) {
this.subscriptions[i].subscription.unsubscribe();
this.subscriptions.splice(i, 1);
i--;
}
}
}
/* 取消订阅 By Id */
public stopSubBySubscriptionId(id: string): void {
for (let i = 0; i < this.subscriptions.length; i++) {
if (id == this.subscriptions[i].id) {
this.subscriptions[i].subscription.unsubscribe();
this.subscriptions.splice(i, 1);
i--;
}
}
}
/* 销毁 */
public destroy(): void {
this.destroyed = true;
for (let i = 0; i < this.subscriptions.length; i++) {
this.subscriptions[i].subscription.unsubscribe();
}
for (let subject of this.subjects.values()) {
subject.unsubscribe();
}
this.subjects.clear();
}
private getSubscriptionId(): string {
let id = Random.generateCharMixed(20);
for (let i = 0; i < this.subscriptions.length; i++) {
if (id == this.subscriptions[i].id) {
this.getSubscriptionId();
}
}
return id;
}
}
其它
可以看到上图代码目录中，rx目录下有一个topic.ts文件。它的作用很简单，就是一个枚举，用来存放各个Topic。
export enum Topic {
// 在这里定义各个Topic
MY_TP_1,
MY_TP_2,
TEXT_TOPIC,
}
具体使用
现在基于Topic的发布订阅的服务端与客户端设计完成，我们就来开始使用它吧。
基础功能
首先是依赖注入。服务端是全局单例的，故需要服务端在根组件中提供出来，也就是让app.component.tsx来提供InnerMqService。
在React中这一点很好办，使用createContext配合Provider即可。
app.component.tsx
const innerMqService = new InnerMqService();
export const GLOBAL = createContext<{ innerMqService: InnerMqService }>({
innerMqService: innerMqService,
});
const AppComponent: React.FC = () => {
return (
<GLOBAL.Provider value={{ innerMqService: innerMqService }}>
<div className="container-app">
<div className="name-app">
app component
</div>
<div className="comp-index">
<IndexComponent></IndexComponent>
</div>
<div className="comp-main">
<MainComponent></MainComponent>
</div>
</div>
</GLOBAL.Provider>
);
}
export default AppComponent;
此处提供的InnerMqService，可以被后代组件通过useContext()实现依赖注入。此部分完成后，就可以在各个组件中去使用了。这里我们需要实现test-a与test-b组件之间的通信，故在这两个组件中去使用它。
以test-a组件为例，将test-a组件当做一个客户端，给定一个Topic，来接收消息：
test-a.component.tsx
const TestAComponent: React.FC = () => {
const global = useContext(GLOBAL);
const client = useRef<InnerMqClient>();
useEffect(() => {
client.current = global.innerMqService.createClient();
client.current.sub(Topic.MY_TP_1, (res) => {
console.log(res);
});
return () => {
client.current && global.innerMqService.destroyClient(client.current);
console.log('test-a销毁客户端');
};
}, []);
return (
<div className="container-test-a">
<div className="name-test-a">
test-a component
</div>
</div>
);
}
export default TestAComponent;
useContext(GLOBAL)的作用即为注入一个由祖先组件或整个应用提供的值，也就是对应上述在根组件中的提供的innerMqService<GLOBAL.Provider value={{ innerMqService: innerMqService }}>。
现在test-a订阅了MY_TP_1这个Topic，现在在test-b中放置一个按钮，使用该Topic给服务端发送消息。只发消息，不做订阅，就不需要客户端了，直接给服务端发送即可。这里同样要将InnerMqService注入到test-b组件：
test-b.component.tsx
const TestBComponent: React.FC = () => {
const global = useContext(GLOBAL);
return (
<div className="container-test-b">
<div className="name-test-b">
test-b component
</div>
<button className="send-btn" onClick={() => {
global.innerMqService.pub(Topic.MY_TP_1, '来自test-b的消息');
}}>
发送
</button>
</div>
);
}
export default TestBComponent;
可以看到，当我点击test-b中的发送按钮，在test-a组件中顺利收到消息了。如果你阅读了前两篇文章，你会发现：除了依赖注入方法不同，具体使用方法是一模一样的。
接下来做个复杂的，在sub-component中放置一个按钮，用于控制test-a的显示与隐藏，同时test-b中改为循环发送：
test-b.component.tsx
const TestBComponent: React.FC = () => {
const global = useContext(GLOBAL);
useEffect(() => {
setInterval(() => {
global.innerMqService.pub(Topic.MY_TP_1, '来自test-b的消息，循环发送');
}, 1000);
}, []);
return (
<div className="container-test-b">
<div className="name-test-b">
test-b component
</div>
<button className="send-btn" onClick={() => {
global.innerMqService.pub(Topic.MY_TP_1, '来自test-b的消息');
}}>
发送
</button>
</div>
);
}
export default TestBComponent;
sub.component.tsx
const SubComponent: React.FC = () => {
const [isShow, setIsShow] = useState(true);
return (
<div className="container-sub">
<div className="name-sub">
sub component
</div>
<div className="comp-test-a">
{isShow ? (<TestAComponent></TestAComponent>) : ''}
</div>
<button className="control-btn" onClick={() => {
setIsShow(!isShow)
}}>
test-a的显示与隐藏
</button>
</div>
);
}
export default SubComponent;
可以看到，在test-a存在的时候，能接收到消息，在test-a从页面上移除时，在生命周期中将客户端销毁，于是也不会接收到消息了。
进阶使用
上文讲解了如何使用inner-mq，本小段中就不再放各个文件的详细代码了，我只描述场景与效果，以及关键代码点。
- 在test-a未存在页面上时，test-b发送消息。然后test-a组件才生成显示在页面上。这就是典型的订阅和发布之间时机错过的问题，一个发布发生在订阅之前的场景。这时候就需要使用之前提到的持久化功能：
test-b.component.tsx的按钮点击事件
<button className="send-btn" onClick={() => {
// persistent: true 表示该消息要做持久化
// ON_ONCE_SUB 表示该消息只进行一次缓存，一次sub后即删除
global.innerMqService.pub(Topic.MY_TP_1, '来自test-b的消息', {
persistent: true,
type: PersistentType.ON_ONCE_SUB
});
}}>
发送
</button>
可以看到，test-b发给test-a的消息，一个都没落下。
- 有时候需要某条消息在每一次订阅的时候发送一遍。同样持久化功能也实现了这一点：
- 当然，基于Topic的发布订阅模式，肯定是支持一对多的，即一个发布者发送消息，所有订阅该Topic的客户端都会收到该条消息。我给每个组件都加上订阅该Topic的代码，可以看到点击test-b中的发送后，每个组件中的订阅都能正确收到消息：
如何在自己项目中使用
很简单，我写工具写插件一向不喜欢发npm，我就喜欢最简单直接的，给源码，自己放到项目中直接调用即可。我认为这样的好处就是不会影响现有依赖，以及我直接给你源码，你可以直接学习并使用，更可以自己定制。
和上次一样，本文给了git地址，直接拉取代码，将里面的rx文件夹拖到自己项目中，就可以愉快地使用了。
结束语
以上即是我使用了很久的跨组件通信的方法。也许不如现成插件那样成熟，但是我认为它降低了理解难度，也更加的解耦了。具体代码已经提交git，大家可以拉取下来自己玩耍，自己设计一些场景来试试。
下一篇讲Java，当然不是在Java中使用RxJS。我会将这三篇文章中跨组件通信的思路从RxJS移植到RxJava上，在Java中分别以Swing桌面与SpringBoot为例，使用与之相同的方法实现跨组件通信。
本文代码git地址：Crimson/rxjs-cross-component-communication (gitee.com)