Websocket解析及用法（封装一个通用订阅发布主题的webSocket类）_websocket 订阅

1、什么是WebSocket?

websocket的目标是通过一个长连接实现与服务器全双工，双向的通信。是一种在单个TCP连接上进行全双工通信的协议，使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在 js中创建websocket时，一个http请求会发送到服务器初始化连接，服务器响应后，连接使用http的upgrade头部从http协议切换到websocket协议，这意味着websocket不能通过标准的http服务器完成，而必须使用支持websocket协议的专有服务器。因为websocket使用了自定义协议，所以url方案稍有变化，不能再使用http://或https://，而要使用我是：ws://和wss://。前者是不安全的连接，后者是安全的。

2、怎么理解WebSocket，可以用在什么场景?

首先，我们思考一个问题，当前端需要展示实时数据，或实现实时聊天功能，我们怎么设计数据请求方案？
按照传统的解决方法就是利用轮询和长轮询的方案，现在我们来对比一下这三种方案的异同：

3、WebSocket对象详解

3-1、websocket()
构造函器会返回一个 WebSocket 对象。
var aWebSocket = new WebSocket(url [, protocols]);

3-2、api
WebSocket.close() 关闭 WebSocket 连接或连接尝试（如果有的话）。如果连接已经关闭，则此方法不执行任何操作;

WebSocket.send() 将需要通过 WebSocket 链接传输至服务器的数据排入队列，并根据所需要传输的 data bytes 的大小来增加 bufferedAmount的值。若数据无法传输（例如数据需要缓存而缓冲区已满）时，套接字会自行关闭。

3-3、属性
WebSocket.binaryType：返回 websocket 连接所传输二进制数据的类型。

WebSocket.bufferedAmount：一个只读属性，用于返回已经被send()方法放入队列中但还没有被发送到网络中的数据的字节数。一旦队列中的所有数据被发送至网络，则该属性值将被重置为 0。但是，若在发送过程中连接被关闭，则属性值不会重置为 0。如果你不断地调用send()，则该属性值会持续增长。

WebSocket.extensions：是只读属性，返回服务器已选择的扩展值。目前，链接可以协定的扩展值只有空字符串或者一个扩展列表。

WebSocket.protocol 是个只读属性，用于返回服务器端选中的子协议的名字；这是一个在创建 WebSocket 对象时，在参数 protocols 中指定的字符串，当没有已建立的链接时为空串。

WebSocket.readyState返回当前 WebSocket 的链接状态，只读。

WebSocket.url是一个只读属性，返回值为当构造函数创建WebSocket实例对象时 URL 的绝对路径。

3-4、事件

WebSocket.onclose：属性返回一个事件监听器，这个事件监听器将在 WebSocket 连接的readyState 变为 3时被调用，它接收一个名字为“close”的 CloseEvent 事件。

WebSocket.onerror：当websocket的连接由于一些错误事件的发生 (例如无法发送一些数据) 而被关闭时，一个error事件将被引发。

WebSocket.onmessage：message 事件会在 WebSocket 接收到新消息时被触发。

WebSocket.onopen：属性定义一个事件处理程序，当WebSocket 的连接状态readyState 变为1时调用;这意味着当前连接已经准备好发送和接受数据。

4、怎么保持WebSocket长时间连接，不断线？

首先我们来探讨一下为什么websocket断线的原因有哪些，由于websocket连接是基于一个tcp连接上的长时间全双工双向通信，所以断线的原因有下几种可能（个人猜测）：
1、网络连接：网络连接不稳定或出现故障，可能会导致WebSocket断线；
2、服务器：WebSocket服务器出现故障或过载，可能会导致WebSocket断线；
3、防火墙问题：防火墙可能会阻止WebSocketClient与服务器之间的通信，导致断线。
4、超时问题：WebSocket在一段时间内没有收到服务器发送的数据，则可能会断开连接。
5、客户端异常中断。

如何解决断线问题？
1、心跳检测机制：通过定期发送心跳消息，可以确保客户端和服务器之间的连接处于活跃状态。如果一段时间内未收到来自客户端的心跳消息，服务器可以认为客户端已经断线，并采取相应的措施，例如关闭连接或重新建立连接。
2、连接状态管理：在客户端代码中维护连接的状态信息，以便及时检测连接的断开和重新连接的状态变化。这样可以使应用程序更容易地处理连接断开和重新连接时的逻辑。
3、异常处理 ：及时捕获和处理在连接过程中可能出现的异常情况，例如网络超时、连接被拒绝等，以提高系统的稳定性和可靠性。

心跳检测机制：
之所以称之为心跳检测是因为：它像心跳一样每隔固定时间发一次，以此来告诉服务器，这个客户端还活着。实际上，这种机制是为了维持长连接而设计的。通常情况下，心跳包的内容并没有特别的规定，通常是一个很小的数据包，甚至可能只包含包头的空包。

心跳检测步骤：
1、客户端会定期发送一个探测包（心跳包）给服务器，以确保连接的活跃性。
2、在发送探测包的同时，客户端会启动一个超时定时器，以便在规定的时间内等待服务器的响应。
3、当服务器接收到客户端发送的探测包时，会立即回应一个确认包，以表明服务器正常接收到了探测包。
4、如果客户端收到服务器的应答包，则说明服务器正常：客户端在收到服务器的确认包后，会立即删除之前启动的超时定时器，表明服务器正常运行。
5、如果客户端的超时定时器超时，仍未收到应答包，则说明服务器挂了：如果客户端的超时定时器到期时仍未收到服务器的确认包，则客户端会认为服务器已经挂了，进而采取相应的措施，例如重新连接或者进行错误处理。

5、封装一个通用的推送、订阅主题的webSocket的类

// 先定义一个基础webSocket类
class WebsocketClient {
  // 连接的URL
  _host = "";
  // websocket对象
  ws = null;
  // 传入的websocket的方法函数，事件回调函数
  _insOwner = null;
  constructor(host: string, ins) {
    this._host = host;
    this._insOwner = ins;
    this.initWebsocket();
  }
  // 初始化创建websocket对象
  initWebsocket() {
    if (this.ws != null) {
      this.ws.close();
    }
    try {
      this.ws = new WebSocket(this._host);
    } catch (error) {
      console.error("create websocket error: ", error.toString());
      if (this._insOwner && typeof this._insOwner.onDisconnect == "function") {
        // websocket连接失败处理函数
        this._insOwner.onDisconnect();
      }
      return;
    }

    this.ws.onopen = () => {
      //开始连接
      if (this._insOwner && typeof this._insOwner.onConnect == "function") {
        this._insOwner.onConnect();
      }
    };
    this.ws.onerror = (error) => {
      console.error("websocket error: ", error);
    };
    this.ws.onclose = (e) => {
      console.error("websocket onclose: ", e);
      if (this._insOwner && typeof this._insOwner.onDisconnect == "function") {
        this._insOwner.onDisconnect(e);
      }
    };
    this.ws.onmessage = (message) => {
      if (this._insOwner && typeof this._insOwner.onMessage == "function") {
        this._insOwner.onMessage(message.data);
      }
    };
  }
  // 发送消息
  sendMessage(message) {
    console.log(this.ws, this.ws.readyState, message);
    //  WebSocket 是否处于打开状态 readyState === 1
    //CONNECTING：值为0，表示正在连接。
    //OPEN：值为1，表示连接成功，可以通信了。
    //CLOSING：值为2，表示连接正在关闭。
    //CLOSED：值为3，表示连接已经关闭，或者打开连接失败。
    if (this.ws && this.ws.readyState == 1) {
      this.ws.send(message);
      return true;
    }
    return false;
  }
  // 关闭连接
  dispose() {
    this._host = "";
    this._insOwner = null;
    if (this.ws != null) {
      this.ws.close();
      this.ws = null;
    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

在此基础类之上定义一个可发布，订阅消息主题的类
什么叫消息主题？在真实开发中，我们接收websocket信息时，后端会在一个ws连接里推送不同页面的信息，这个时候我们就要判断当前推送的信息是不是当前页面要用的信息，如果是就调用接收信息的回调，不是就忽略。示例如下：

可发布，订阅消息主题的类代码如下
1

interface UpcWebsocketEvent {
  type: UpcWebsocketEventType;
  data: UpcWebsocketEventInfo;
}

interface UpcWebsocketEventInfo {
  isReconnect: boolean; // 是否是重连
  reconnectable: boolean; // 是否自动重连
  reconnectDivide: number; // 重连间隔 S
  reconnectMaxTimes?: number; // 最大重连次数
  reconnectTimes?: number; // 当前重连次数
}

export enum UpcWebsocketEventType {
  Connected,
  Disconnect,
  Reconnect,
}

export class UpcWebsocketClient {
  _client: any = null; // 存储WebSocket客户端实例
  _url = "";
  _urlGenerator = null; // url 生成器

  isConnect = false; //连接状态
  reConnectNum = 0; //重连次数
  reConnectMaxNum = 3; // 最大重连次数

  // 登录相关
  _logined = false;
  _loginTimer = null;
  _loginResolve = null;

  // 心跳相关
  _heartTimer = null;
  _heartBeatResolve = null;
  _heartBeatResolveTimer = null;

  // 重连句柄
  _autoReloginTimer = null;
  _autoReloginDivide = 10; // 每隔 10 S 发起一次重连

  // 实例事件, 需全局订阅
  _instanceEventCallback: ((event: UpcWebsocketEvent) => void) | null = null;
  // 主题订阅
  _subIndex = 0;
  _subMap = new Map<string, Map<number, (data: any) => void>>(); // Map<topic, Map<subIndex, listner>>

  constructor(urlGenerator) {
    if (typeof urlGenerator != "function") {
      throw new Error("urlGenerator is not function");
    }
    this._urlGenerator = urlGenerator;
  }

  onInstanceEvent(fn: (event: UpcWebsocketEvent) => void) {
    if (typeof fn == "function") {
      this._instanceEventCallback = fn;
    }
  }

  emitInstanceEvent(type: UpcWebsocketEventType, data: UpcWebsocketEventInfo) {
    if (typeof this._instanceEventCallback == "function") {
      this._instanceEventCallback({ type, data });
    }
  }
// 根据基础类创建websocket对象
  async createWebsocket() {
    // this._client存储创建出来的websocket对象索引
    if (this._client != null) {
      this._client.dispose();
      this._client = null;
    }
    // 拿到websocket连接url,一般都是要携带token的，所以拿url的方法要抽离出类之外，通过构造函数传入
    const result = await this._urlGenerator();
    if (result.code != 0) {
      return { code: result.code, msg: `获取推送服务认证口令失败` };
    }
    this._url = result.data; //    `${this._host}/ws?token=${this._token}`;

    return new Promise((resolve: (connected: boolean) => void) => {
      this._loginResolve = resolve;
      this._client = new WebsocketClient(this._url, this);
    });
  }
  // 开始连接，如果连接失败，将尝试自动重连，开启连接
  async start() {
    const connected = await this.createWebsocket();
    // 判断ws对象是否创建成功，否就重连创建，是就开启心跳检测机制
    if (!connected) {
      this.autoRelogin();
    } else {
      this.keepHeartBeat(); // 开始心跳
    }
  }
  // 暂停连接，清理相关资源
  pause() {
    // 清理 socket
    this._client && this._client.dispose();
    this._client = null;

    // 如果有心跳，取消心跳
    this._heartTimer && clearTimeout(this._heartTimer);
    this._heartTimer = null;

    // 如果心跳正在进行, 取消回调
    this._heartBeatResolveTimer && clearTimeout(this._heartBeatResolveTimer);
    this._heartBeatResolveTimer = null;
    this._heartBeatResolve = null;

    this._loginResolve = null;

    // 如果在重连, 去掉重连
    this._autoReloginTimer && clearTimeout(this._autoReloginTimer);
    this._autoReloginTimer = null;

    // 清理主题订阅
    this._subIndex = 0;
    this._subMap.clear();

    this.isConnect = false; //连接状态
    this.reConnectNum = 0;

    return;
  }
  // 重连,每隔一定时间尝试重新连接
  async autoRelogin() {
    this._autoReloginTimer = setTimeout(async () => {
      this._autoReloginTimer = null;
      const ret = await this.createWebsocket();
      if (!ret) {
        return this.autoRelogin();
      } else {
        this.keepHeartBeat(); // 开始心跳
      }
      this.isConnect = true;
      // 重连成功
      this.emitInstanceEvent(UpcWebsocketEventType.Reconnect, {
        msg: "",
        relogin_divide: this._autoReloginDivide,
      });
    }, this._autoReloginDivide * 1000);
  }
  // 建立连接，连接建立成功之后要做的事情，基础类的onopen事件回调函数
  onConnect() {
    console.log(`webSocket已连接`);
    this.isConnect = true;
    this.reConnectNum = 0;
    typeof this._loginResolve == "function" && this._loginResolve(true);
    this._loginResolve = null;
    this.recoverSubscribe(); // 恢复订阅
  }
  // WebSocket连接断开时调用，清理资源，基础类的onclose事件回调函数
  onDisconnect(e?) {
    console.log(`webSocket已经关闭 ${e} `);
    // 清理 socket
    this._client && this._client.dispose();
    this._client = null;

    // 如果有心跳，取消心跳
    this._heartTimer && clearTimeout(this._heartTimer);
    this._heartTimer = null;
    // 如果心跳正在进行, 取消回调
    this._heartBeatResolveTimer && clearTimeout(this._heartBeatResolveTimer);
    this._heartBeatResolveTimer = null;
    this._heartBeatResolve = null;

    // 触发掉线通知掉线
    this.emitInstanceEvent(UpcWebsocketEventType.Disconnect, {
      msg: "",
      relogin_divide: this._autoReloginDivide,
    });

    //被动断开，重新连接
    if (e && e?.code) {
      this.autoRelogin();
      console.log("websocket连接不上，请重新登录或联系开发人员!");
    }
  }
  // 接收到的 WebSocket消息，解析消息并调用相应的处理函数，基础类的onmessage事件回调函数
  onMessage(message: string) {
    const msg = JSON.parse(message);
    // console.log("onMessage", msg);
    if (!msg.channel) {
      console.error("onMessage error ");
      return;
    }
    if (
      !(typeof msg.text == "string" && JSON.parse(msg.text).action_ans == "subscriber successed!")
    ) {
      this.handlePublish(msg);
    }
  }

  // 使用nginx代理webSocket链接，客户端和服务器握手成功后，如果在60s时间内没有数据交互，连接就会自动断开
  // 心跳检测，定期 30s 发送心跳请求并处理心跳响应
  keepHeartBeat() {
    this._heartTimer = setTimeout(async () => {
      this._heartTimer = null;
      const ret = await this.sendHeartBeat();
      // if (ret?.code == 1) {
      //   return this.onDisconnect();
      // }
      return this.keepHeartBeat();
    }, 30 * 1000); // 
  }
  // 发送心跳请求，返回一个Promise，用于处理心跳响应
  sendHeartBeat() {
    return new Promise((resolve) => {
      this._heartBeatResolve = resolve;
      this._heartBeatResolveTimer = setTimeout(() => {
        this._heartBeatResolveTimer = null;
        if (this._heartBeatResolve !== null) {
          this._heartBeatResolve({ code: 100, ret_msg: "sendHeartBeat timeout", data: null });
          this._heartBeatResolve = null;
        }
      }, 10 * 1000);
      const reqData = {
        action: "req",
        channel: "ping",
      };
      this._client && this._client.sendMessage(JSON.stringify(reqData));
    });
  }
  // 
  handlePublish(msg) {
    const topic = msg.channel;
    const handleMap = this._subMap.get(topic);
    if (handleMap != null) {
      handleMap.forEach((fn) => {
        try {
          fn(JSON.parse(msg.text));
        } catch (e) {
          console.log(e);
        }
      });
    }
  }
  // 恢复所有主题的订阅
  recoverSubscribe() {
    this._subMap.forEach((handleMap, topic) => {
      if (handleMap != null && handleMap.size > 0) {
        const reqData = {
          action: "sub",
          channel: topic,
          data: {},
        };
        this._client && this._client.sendMessage(JSON.stringify(reqData));
      }
    });
  }
  /**
   * 订阅指定主题，并注册事件监听器。
   * @param topic 订阅的主题字符串。
   * @param listener 当订阅主题触发时执行的回调函数。
   * @returns 返回一个句柄，用于取消订阅。
   */
  subscribe(topic: string, listener: () => void) {
    if (typeof listener != "function") return;
    let handleMap = this._subMap.get(topic);
    if (handleMap == null) {
      handleMap = new Map();
      this._subMap.set(topic, handleMap);
    }
    const handle = ++this._subIndex;
    handleMap.set(handle, listener);
    if (handleMap.size == 1) {
      // 初始添加订阅, 需要向后台发送消息
      const reqData = {
        action: "sub",
        channel: topic,
        data: {},
      };
      this._client && this._client.sendMessage(JSON.stringify(reqData));
    }
    return handle;
  }
  /**
   * 取消订阅指定主题或删除指定句柄对应的监听器。
   * @param topic 要取消订阅的主题字符串。
   * @param handle 可选，取消特定句柄对应的监听器；若未提供，则取消该主题下的所有监听器。
   */
  unsubscribe(topic, handle?) {
    const handleMap = this._subMap.get(topic); // 获取对应主题的订阅
    if (handleMap == null || handleMap.size == 0) {
      // 未订阅，无需退订
      return;
    }
    if (typeof handle == "undefined") {
      // 全部退订
      handleMap.clear();
    } else {
      handleMap.delete(handle);
    }
    if (handleMap.size == 0) {
      // 向后台取消订阅
      const reqData = {
        action: "unsub",
        channel: topic,
        data: {},
      };
      this._client && this._client.sendMessage(JSON.stringify(reqData));
    }
  }
  /**
   * 清理资源并释放所有内部状态。
   */
  dispose() {
    // 清理 socket
    this._client && this._client.dispose();
    this._client = null;

    // 如果有心跳，取消心跳
    this._heartTimer && clearTimeout(this._heartTimer);
    this._heartTimer = null;

    // 如果心跳正在进行, 取消回调
    this._heartBeatResolveTimer && clearTimeout(this._heartBeatResolveTimer);
    this._heartBeatResolveTimer = null;
    this._heartBeatResolve = null;

    // 如果在重连, 去掉重连
    this._autoReloginTimer && clearTimeout(this._autoReloginTimer);
    this._autoReloginTimer = null;

    // 清理系统订阅
    this._instanceEventCallback = null;
    // 清理主题订阅
    this._subIndex = 0;
    this._subMap.clear();
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332