前端实时数据传输的需求越来越多，而 WebSocket 是一种比较流行的解决方案。本文将介绍如何使用 RxJS 来简化 WebSocket 消息推送的过程。
什么是 RxJS？
RxJS 是一个响应式编程库，可以让开发者用简单的代码处理异步事件流。它提供了一个函数式编程风格的 API，让开发者可以用 clean、简洁的方式处理数据流。它拥有广泛的应用场景，其中包括 WebSocket 消息推送。
实现 WebSocket 消息推送的步骤
在介绍 RxJS 如何实现 WebSocket 消息推送之前，先看一下实现 WebSocket 消息推送的步骤：
- 创建一个 WebSocket 实例，指定 WebSocket 服务器的地址
- 监听 WebSocket 实例的连接状态，如果连接成功，就可以向服务器发送消息了
- 监听 WebSocket 实例的消息事件，处理接收到的消息
使用原生的 WebSocket API 来实现这些步骤比较繁琐，但是使用 RxJS 就简单多了。
首先我们需要安装 RxJS：
npm install rxjs --save
接下来，我们来一步步实现 WebSocket 消息推送：
创建一个 WebSocket 的 Observable
我们可以使用 RxJS 中的 Observable 来创建一个 WebSocket 的 Observable，它表示一组未来可能发生的事件。我们只需要在 Observable.create() 函数中创建一个 WebSocket 实例，并且在 open 回调函数中发出连接成功的事件，即可创建一个 WebSocket 的 Observable。代码如下：
-- -------------------- ---- -------
------ - ---------- - ---- -------
----- ------------------------- - --- -- -
----- -- - --- ---------------
----- ---------- - -------------------------- -- -
--------- - ----- -- -
--------------- ----- ------- ----- ----- ---
--
------------ - ----- -- -
--------------- ----- ---------- ----- ----- ---
--
---------- - ----- -- -
----------------------
--
---------- - -- -- -
--------------------
--
------ -- -- -
-----------
--
---
------ -----------
--这个函数接收一个 WebSocket 的地址 url，并且返回一个 WebSocket 的 Observable。在 Observable.create() 函数中，我们通过 ws.onopen、ws.onmessage、ws.onerror、ws.onclose 四个 WebSocket 的事件来发出 Observable 中的事件:
- 当 WebSocket 成功连接时，
ws.onopen回调函数中会调用observer.next({ type: 'open', data: event })，发出一个类型为open的事件，并且将event数据作为参数传递过去。 - 当 WebSocket 接收到消息时，
ws.onmessage回调函数中会调用observer.next({ type: 'message', data: event })，发出一个类型为message的事件，并且将event数据作为参数传递过去。 - 当 WebSocket 发生错误时，
ws.onerror回调函数中会调用observer.error(event)，发出一个错误型事件，并且将event数据作为参数传递过去。 - 当 WebSocket 关闭时，
ws.onclose回调函数中会调用observer.complete()，发出一个完成事件，表示 WebSocket 已经关闭。
处理事件流
我们使用 createWebSocketObservable() 函数创建了 WebSocket 的 Observable，但是我们并不知道如何用它来推送消息。在使用 RxJS 中的操作符之前，我们需要了解 Observable 中的事件流，以及如何对它进行处理。
RxJS 提供了大量的操作符，可以对事件流进行各种处理，包括 map、filter、merge、concat、flatMap 等等。在这里，我们只需要使用其中两个操作符：filter 和 map。
- 使用
filter()操作符可以过滤掉某些不需要的事件，只处理某些满足条件的事件。例如，当我们只需要处理接收到的消息时，可以这样写：
const messageObservable = createWebSocketObservable('ws://localhost:8080')
.pipe(filter(event => event.type === 'message'));- 使用
map()操作符可以将一个事件流中的事件转化成另一种事件。例如，当我们只需要处理接收到的消息中的数据时，可以这样写：
const messageDataObservable = createWebSocketObservable('ws://localhost:8080')
.pipe(
filter(event => event.type === 'message'),
map(event => event.data),
);订阅事件
通过使用 createWebSocketObservable() 函数，我们得到了 WebSocket 的 Observable，并且通过使用 filter() 和 map() 等操作符，我们可以对 WebSocket 的事件流进行处理。接下来，我们需要使用 subscribe() 函数订阅事件流，才能真正地推送消息。例如，当我们需要在控制台中输出接收到的消息的内容时，可以这样写：
const subscription = messageDataObservable.subscribe(
data => console.log(data),
error => console.error(error),
() => console.log('WebSocket connection closed'),
);在这个例子中，我们使用 subscribe() 函数订阅了 messageDataObservable 中的事件流，并且传入了三个回调函数：
- 当
messageDataObservable中发出一个next事件时，会调用data => console.log(data)回调函数，将事件的数据作为参数打印出来。 - 当
messageDataObservable中发出一个error事件时，会调用error => console.error(error)回调函数，将事件的错误信息作为参数打印出来。 - 当
messageDataObservable中发出一个complete事件时，会调用() => console.log('WebSocket connection closed')回调函数。
推送消息
订阅事件之后，我们就可以将消息推送到 WebSocket 服务器了。推送消息只需要使用 WebSocket 的 send() 函数，将消息发送到服务器即可。例如，当我们想通过 WebSocket 发送一个消息时，可以这样写：
-- -------------------- ---- -------
------------------------------------------------
-----------
-- -- ---------------------- ---------- --------------
----- -- ---------------------
-- -- ---------------------- ---------- ---------
--
----- ------- - -----------------------------------
----- ----- - ---------------------------------
--------------------------------- -- -- -
----- ------- - ------------
-----------------
---在这个例子中，我们通过 createWebSocketObservable() 函数创建了一个 WebSocket 的 Observable，并且在订阅事件时打印了连接状态。
我们还定义了一个点击按钮 sendBtn，当点击按钮时，将 input 中的内容作为消息，使用 WebSocket 的 send() 函数将消息发送给服务器。
总结
使用 RxJS 可以很方便地处理 WebSocket 消息推送，将复杂的步骤简化成几行代码。在实际项目中，RxJS 还有很多更加强大的操作符，可以帮助我们更好地处理各种异步事件流。
来源：JavaScript中文网 ，转载请注明来源 https://www.javascriptcn.com/post/64f483a4f6b2d6eab3d7e93f