前言
在现代 Web 应用中，实时数据交互已经成为了许多场景的必要需求，而 Websocket 作为一种流行的实现方式，可以使用它来处理实时数据交互的需求。而使用 AngularJS 和 RxJS 的组合，我们可以快速、简单地实现 Websocket 实时数据交互。
RxJS 简介
RxJS 是 ReactiveX 的 JavaScript 版本，它是一种编写异步和基于事件的代码的工具。RxJS 提供了丰富的函数式编程工具，使得处理并发、异步事件和数据流变得非常简单。
RxJS 通过 Observable 对象封装和处理不同的事件和数据流。Observable 可以组合和转化，这样我们就可以编写链式函数式编程代码。
创建 Websocket 连接
首先，我们需要使用 RxJS 创建一个 Websocket 的连接。可以使用 Rx.Observable.fromEvent 创建一个 Observable 对象，来监听 Websocket 的 'open' 事件，从而创建 Websocket 连接：
-- -------------------- ---- -------
------ - ---------- - ---- ------------------
----- ---------- --------- - --- ---------------------------------
----- ---------------- - ------------------------------- --------
----------------
------------- -- -
---------------------- ---------- -------
---在这个例子中，使用 Rx.Observable.fromEvent 方法来创建一个 Observable 对象，监听 Websocket 的打开事件。然后使用 subscribe 方法，将回调函数绑定到 Observable 对象上，以便在 Websocket 连接打开时执行该回调函数。
发送和接收消息
一旦建立了 Websocket 连接，我们就可以发送和接收消息了。我们可以使用 Websocket 的 send 方法和 RxJS 的 fromEvent 方法来实现该功能。
这里顺带说一下，当我们使用 send 方法发送消息时，消息会异步地发送到 Websocket，这意味着我们需要等待 Websocket 的响应，以确定消息是否已成功发送到对端。假如我们需要一系列的异步操作完成后再执行下一步操作，这时我们可以使用 RxJS 的 merge 方法来合并多个 Observable 对象，实现并行执行多个异步操作。
-- -------------------- ---- -------
----- ------- - ------ --------
------------------------
----- ------------- - ------------------------------- ----------
------------ ------------- -- ------------
-------------
----------------- ---- -- -
--------------------- ---------- ------
---在这个例子中，我们首先使用 Websocket 的 send 方法发送一个消息。然后使用 RxJS 的 fromEvent 方法来创建一个 Observable 对象，监听从 Websocket 接收到的消息。这里通过 map 操作符将 MessageEvent 对象转化为消息内容，最后将接收到的消息打印到控制台。
处理异常和关闭连接
在使用 Websocket 进行实时数据交互时，经常要处理连接关闭或发生异常的情况。这时我们可以使用 RxJS 的 catchError 和 finalize 方法，来处理 Websocket 关闭或出现异常的情况。
-- -------------------- ---- -------
----- ---------------- - ------------------------------- --------
----- ----------------- - ------------------------------- --------
----------- -- -------------------- ---------------- ---------- -----------
----- ----------------- - ------------------------------- --------
---------------- ---- -- -------------------- ---------------- ----------- ------ ---------------------
----- ------------- - ------------------------------- ----------
------------ ------------- -- ------------
----------------
------------- -- -
---------------------- ---------- -------
---
------------------------ ------------------
------
------------------ ---- -- -
---------------------
------ ---------
---
----------- -- -
---------------------- ---------- ---------
---
-
------------- -- -
-- -- --------- -- ---------- ------ -- -----
---在这个例子中，我们使用 fromEvent 方法监听 Websocket 的 'close' 和 'error' 事件，并使用 flatMap 和 throw 方法来创建一个错误 Observable 对象。在 Observable 对象最后的 subscribe 中，使用 catchError 方法来处理抛出的错误，最后使用 finalize 方法在 Websocket 关闭时执行清理任务。
总结
使用 AngularJS 和 RxJS 的组合可以非常方便地实现 Websocket 的实时数据交互。同时 RxJS 提供了丰富的函数式编程工具，可以大大简化异步编程和数据流处理的复杂度，从而使得代码更加简单、易于维护。
完整示例代码请见 Github。
致谢
本文的示例代码参考了 Angular 4 使用Rx.js实现WebSocket实时信息推送。
来源：JavaScript中文网 ，转载请注明来源 https://www.javascriptcn.com/post/65a1dca1add4f0e0ff9f7ed3