在线问答5

这个屌丝很懒，什么也没留下！

热门标签

React+WebSocket(SSE)实时通知+语音播报+websocket stomp协议_react sse

作者：在线问答5 | 2024-06-30 00:15:22

踩

react sse

文章目录

前言
一、实时通知用哪个？
二、SSE尝试
- 1.公共组件
- 2.联调测试
三、WebSocket尝试
四、stomp协议尝试
总结

前言

大概内容：

随着需求版本迭代更新，新增系统消息语音播报通知，本文就介绍了前端实时获取后台消息的内容。

提示：以下是本篇文章正文内容，下面案例可供参考

一、实时通知用哪个？

对比如下图
各通信技术对比
1、sass系统轮询会加重服务器压力，故不能使用该方式。
2、同为浏览器推送技术，相较于 WebSocket 而言，Server-Sent Events （简称SSE）更少被人知晓，具体实践也较少。
原因有两点：

WebSocket 比 SSE 更强大，Websocket 在客户端和服务器之间建立了双向的实时通信。而 SSE 只支持从服务器到客户端的单向实时通信。
WebSocket 在浏览器方面支持更广（详见下图），IE / Edge 几乎根本不支持 SSE

然而，就第一点而言，与 WebSocket 相比，SSE 也有独特的优势
SSE 的浏览器端实现内置断线重连和消息追踪的功能，WebSocket 也能实现，但是不在协议设计范围内，需要手动处理。
SSE 实现简单，完全复用现有的 HTTP 协议，而 WebSocket 是相对独立于 HTTP 的一套标准，跨平台实现较为复杂。

二、SSE尝试

1.公共组件

代码如下（示例）：

import React, { Component } from 'react';
import { Button, Comment, Tooltip, Avatar, Steps, notification } from 'antd';
import moment from 'moment';
import { Fetch } from 'kit';
const openNotification = (type, msg) => {
  notification[type]({
    message: '系统消息',
    description: msg,
    duration: 5
  });
};

export default class Sse extends Component {
  state = {
    current: 0,
    status: 'wait' as any,
    systemMessages: [{ msg: '【hello ypf】 请创建连接以获取后台消息！' }]
  };
  componentDidMount() {
    const token = (window as any).token;
    if (token) {
      this.createSseConnect();
    }
  }

  // 新建Sse连接
  createSseConnect = () => {
    if (window.EventSource) {
      var source = new EventSource(
        `http://localhost:3000/sse/createSseConnect?clientId=123`,
        { withCredentials: true }
      );

      // 监听打开事件
      source.addEventListener('open', (e) => {
        console.log('打开连接 onopen==>', e);
        this.setState({ current: 1, status: 'process' });
        openNotification('success', '建立连接成功');
      });

      // 监听消息事件
      source.addEventListener('message', (e) => {
        let systemMessages = this.state.systemMessages;
        const data = JSON.parse(e.data);
        const code = data[0].code;
        const msg = data[0].msg;

        if (code === 200) {
          openNotification('success', msg);
          systemMessages.push({ msg: msg });
          this.setState({ systemMessages: systemMessages });
        } else if (code === 0) {
          // 然后状态码为000 把客户端id储存在本地
          sessionStorage.setItem('clientId', msg);
        }
        console.log(systemMessages);
      });

      // 监听错误事件
      source.addEventListener('error', (e) => {
        let systemMessages = this.state.systemMessages;
        console.log('错误事件', e);
        openNotification('error', '已断开与后端连接');
        systemMessages.push({ msg: '已断开与后端连接' });

        this.setState({
          current: 0,
          status: 'error',
          systemMessages: systemMessages
        });
      });
      // 关闭连接
      source.close = function() {
        console.log('断开 οnerrοr==>');
      };
    } else {
      alert('该浏览器不支持sse');
    }
  };

  // 获取系统消息
  getSystemMessage = () => {
    // 发送网络请求
    try {
      Fetch('http://localhost:3000/api1/sse/broadcast', {
        method: 'POST'
      }).then((val) => {
        let { res } = val as any;
        if (res.code == '000000') {
          let systemMessages = this.state.systemMessages;
          systemMessages.push({ msg: '参数' });
          this.setState({
            systemMessages: systemMessages
          });
        }
      });
    } catch (error) {
      console.error(error);
    }
  };

  // 断开连接
  closeSseConnect = () => {
    // 先获取到本地存储的clientId 再
    const clientId = sessionStorage.getItem('clientId');
    if (clientId === null) {
      return;
    }

    try {
      Fetch(
        `http://localhost:3000/api1/sse/closeConnect?clientId=${clientId}`,
        {
          method: 'POST'
        }
      ).then((val) => {
        let { res } = val as any;
        if (res.code == '000000') {
          let systemMessages = this.state.systemMessages;
          systemMessages.push({ msg: '已断开与后端连接' });
          this.setState({
            current: 0,
            status: 'error',
            systemMessages: systemMessages
          });
        }
      });
    } catch (error) {
      console.error(error);
    }
  };

  render() {
    const { Step } = Steps;
    const { current, status, systemMessages } = this.state;

    return (
      <div className="center">
        <Button
          style={{ margin: '0px 20px 20px 0px' }}
          type="primary"
          onClick={() => {
            this.createSseConnect();
          }}
        >
          创建连接
        </Button>
        <Button
          type="primary"
          style={{ margin: '0px 20px 20px 0px' }}
          onClick={() => {
            this.getSystemMessage();
          }}
        >
          获取消息
        </Button>
        <Button
          type="danger"
          onClick={() => {
            this.closeSseConnect();
          }}
        >
          断开连接
        </Button>

        <Steps direction="vertical" current={current} status={status}>
          <Step title="成功建立SSE连接" description="successful connected" />
          <Step title="接收后端通知中" description="waiting for message" />
        </Steps>

        {systemMessages.map((systemMessage, index) => {
          return (
            <div
              style={{
                border: '1px solid #ccc',
                borderRadius: '10px',
                margin: '10px auto',
                padding: '0px 15px'
              }}
              key={index}
            >
              <Comment
                key={systemMessage.msg}
                author={<a>系统消息</a>}
                avatar={
                  <Avatar
                    src="https://joeschmoe.io/api/v1/random"
                    alt="Han Solo"
                  />
                }
                content={<p>{systemMessage.msg}</p>}
                datetime={
                  <Tooltip title={moment().format('YYYY-MM-DD HH:mm:ss')}>
                    <span>{moment().fromNow()}</span>
                  </Tooltip>
                }
                className="comment"
              />
            </div>
          );
        })}
      </div>
    );
  }
}

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

2.联调测试

由于后台项目限制局限，不能设置text/event-stream直接返回数据接口（不能直接返回response完整流），故这种方式暂不能成功
原理如下图

SSE 协议很简单，本质上是一个客户端发起的 HTTP Get 请求，服务器在接到该请求后，返回 200 OK 状态，同时附带以下 Headers
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
SSE 的 MIME Type 规定为 text/event-stream
SSE 肯定不允许缓存
SSE 是一个一直打开的 TCP 连接，所以 Connection 为 Keep-Alive之后，服务器保持连接，在 Body 中持续发送文本流，以实现实时消息推送。

三、WebSocket尝试

1.公共组件

在react中使用websocket不需要引入其他库，只需要创建一个公共组件，封装一下websocket
websoketComponent.tsx

import React, { Component } from 'react';
import {
  Button,
  Comment,
  Tooltip,
  Avatar,
  Steps,
  notification,
  message
} from 'antd';
import moment from 'moment';
import { createWebSocket, closeWebSocket, websocket } from './websoket';
const newOrder = require('./images/newOrder.mp3');

const openNotification = (type, msg) => {
  notification[type]({
    message: '系统消息',
    description: msg,
    duration: 5
  });
};

export default class Websoket extends Component {
  state = {
    current: 0,
    status: 'wait' as any,
    systemMessages: [{ msg: '【hello admin】 请创建连接以获取后台消息！' }]
  };
  componentDidMount() {
    const loginInfo = JSON.parse(sessionStorage.getItem('登录信息'));
    if (!loginInfo) {
      return null;
    }
    const isMasterAccount = loginInfo && loginInfo.isMasterAccount;
    //isMasterAccount主账号接收消息
    const token = (window as any).token;
    if (token && isMasterAccount) {
      this.createWebsoketConnect();
    }
  }
  componentWillUnmount() {
    this.setState = (state, callback) => {
      return;
    };
  }

  // 新建Websoket连接
  createWebsoketConnect = (type?) => {
    let Const='https://www.baidu.com'
    const loginInfo = JSON.parse(sessionStorage.getItem('登录信息'));
    const employeeId = loginInfo && loginInfo.employeeId;
    let localarr = Const.split('/'); //所带参数以/进行分开
    console.log('www.baidu.com', localarr[2], '配置域名');

    if (typeof WebSocket != 'undefined') {
      var music = new Audio();
      music.src = newOrder; // 声音文件

      //如果网站使用HTTPS，WebSocket必须要使用wss协议；使用wss协议的连接请求必须只能写域名，而非IP+端口；
      //如果网站使用HTTP，WebSocket必须要使用ws协议；
      let url = `wss://${localarr[2]}/websocket?userId=${employeeId}`; //服务端连接的url
      if (websocket && websocket.readyState == 1) {
        this.setState({ current: 1, status: 'process' });
        return false;
      }
      createWebSocket(url);
      if (websocket) {
        this.setState({ current: 1, status: 'process' });
        if (type == 'manualOperation') {
          let systemMessages = this.state.systemMessages;
          openNotification('success', '建立连接成功');
          systemMessages.push({ msg: '建立连接成功' });
        }
      }

      websocket.onmessage = (event) => {
        if (event.data != 'null' && event.data != 'order') {
          let systemMessages = this.state.systemMessages;
          openNotification('success', event.data);
          systemMessages.push({ msg: event.data });
          this.setState({ systemMessages: systemMessages });
        } else if (event.data == 'order') {
          let autoplay = true;
          music
            .play()
            .then(() => {
              // 支持自动播放
              autoplay = true;
            })
            .catch((err) => {
              // 不支持自动播放
              autoplay = false;
            })
            .finally(() => {
              music.remove();
              // 告诉调用者结果
            });
          if (!autoplay) {
            message.info('不支持支持自动播放');
          }
          let systemMessages = this.state.systemMessages;
          openNotification('success', '您有一笔新的订单请注意查收!');
          systemMessages.push({ msg: '您有一笔新的订单请注意查收!' });
          this.setState({ systemMessages: systemMessages });
        }
      };
      websocket.onclose = (event) => {
        let systemMessages = this.state.systemMessages;
        if (type == 'manualOperation') {
          openNotification('error', '已断开与后端连接');
        }
        systemMessages.push({ msg: '已断开与后端连接' });

        this.setState({
          current: 0,
          status: 'error',
          systemMessages: systemMessages
        });
        this.closeWebsoketConnect();
      };
    } else {
      message.error('您的浏览器不支持WebSocket');
    }
  };

  // 断开连接
  closeWebsoketConnect = () => {
    if (websocket.readyState == 3) {
      console.error('连接已经关闭');
    } else {
      closeWebSocket();
    }
    let systemMessages = this.state.systemMessages;

    systemMessages.push({ msg: '主动已断开与后端连接' });
    this.setState({
      current: 0,
      status: 'wait',
      systemMessages: systemMessages
    });
  };

  render() {
    const { Step } = Steps;
    const { current, status, systemMessages } = this.state;

    return (
      <div className="center">
        <Button
          style={{ margin: '0px 20px 20px 0px' }}
          disabled={current == 1}
          type="primary"
          onClick={() => {
            this.createWebsoketConnect('manualOperation');
          }}
        >
          创建连接
        </Button>

        <Button
          type="danger"
          disabled={current == 0}
          onClick={() => {
            this.closeWebsoketConnect();
          }}
        >
          断开连接
        </Button>

        <Steps direction="vertical" current={current} status={status}>
          <Step title="成功建立SSE连接" description="successful connected" />
          <Step title="接收后端通知中" description="waiting for message" />
        </Steps>

        {systemMessages.map((systemMessage, index) => {
          return (
            <div
              style={{
                border: '1px solid #ccc',
                borderRadius: '10px',
                margin: '10px auto',
                padding: '0px 15px'
              }}
              key={index}
            >
              <Comment
                key={systemMessage.msg}
                author={<a>系统消息</a>}
                avatar={
                  <Avatar
                    src="https://joeschmoe.io/api/v1/random"
                    alt="Han Solo"
                  />
                }
                content={<p>{systemMessage.msg}</p>}
                datetime={
                  <Tooltip title={moment().format('YYYY-MM-DD HH:mm:ss')}>
                    <span>{moment().fromNow()}</span>
                  </Tooltip>
                }
                className="comment"
              />
            </div>
          );
        })}
      </div>
    );
  }
}


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

websoket.ts

let websocket,
  lockReconnect = false;
let createWebSocket = (url) => {
  websocket = new WebSocket(url);
  websocket.onopen = function() {
    heartCheck.reset().start();
  };
  websocket.onerror = function(e) {
    console.log(e, '错误');
    const loginInfo = JSON.parse(sessionStorage.getItem('登录信息'));
    if (!loginInfo) {
      return null;
    }
    //本项目取消重连机制，可以使用
    // reconnect(url);
  };
  websocket.onclose = function(e) {
    console.log(
      'websocket 断开: ' + e.code + ' ' + e.reason + ' ' + e.wasClean
    );
  };
  websocket.onmessage = function(event) {
    lockReconnect = true;
    //event 为服务端传输的消息，在这里可以处理
  };
};
//重连机制
let reconnect = (url) => {
  if (lockReconnect) return;
  //没连接上会一直重连，设置延迟避免请求过多
  setTimeout(function() {
    createWebSocket(url);
    lockReconnect = false;
  }, 5000);
};
//心跳消息
let heartCheck = {
  timeout: 29000, //29秒
  timeoutObj: null,
  reset: function() {
    clearInterval(this.timeoutObj);
    return this;
  },
  start: function() {
    this.timeoutObj = setInterval(function() {
      //这里发送一个心跳，后端收到后，返回一个心跳消息，
      //onmessage拿到返回的心跳就说明连接正常
      if (websocket.readyState != 3) {
        //根据readyState属性可以判断webSocket的连接状态，该属性的值可以是下面几种：
        // 0 ：对应常量CONNECTING (numeric value 0)，
        //  正在建立连接连接，还没有完成。The connection has not yet been established.

        // 1 ：对应常量OPEN (numeric value 1)，
        //  连接成功建立，可以进行通信。The WebSocket connection is established and communication is possible.

        // 2 ：对应常量CLOSING (numeric value 2)
        //  连接正在进行关闭握手，即将关闭。The connection is going through the closing handshake.

        // 3 : 对应常量CLOSED (numeric value 3)
        //  连接已经关闭或者根本没有建立。The connection has been closed or could not be opened.
        websocket.send('HeartBeat');
      }
    }, this.timeout);
  }
};
//关闭连接
let closeWebSocket = () => {
  websocket && websocket.close();
};
export { websocket, createWebSocket, closeWebSocket };

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

找一个地方把组件引入

 <WebsoketComponent />
1

newOrder.mp3
网上随便搜一个就行。

上边代码是进入系统直接连接的例子，开发过程中可以先用一个Modal自测

 <Modal
   visible={this.state.serverMessage}
    onCancel={this.getServerMessage}
    title="服务端消息通知"
    maskClosable={false}
    width={800}
    footer={null}
  >
    <WebsoketComponent />
  </Modal>
1
2
3
4
5
6
7
8
9
10

自测modal

2.遇到问题

播放音频存在问题
谷歌浏览器做了改革，不再允许自动播放音频和视频。Chrome只允许用户对网页进行主动触发后才可自动播放音频和视频。
本项目发现如果浏览器处于最上层位置，就是打开的是当前标签页，音乐是可以播放的，所以需求就通过了。

3.参考链接

阮一峰 Server-Sent Events 教程
 阮一峰 WebSocket 教程
 基于Springboot+SSE+React
websocket实时通信

四、stomp协议尝试

由于后台是微服务集群springboot，所以单节点消费会存在问题，故更换方式

直接上代码了

1、安装sockjs-client和stompjs

npm install sockjs-client
npm install stompjs
1
2

2、引入socket库

import SockJS from 'sockjs-client';
import Stomp from 'stompjs'
1
2

3、新建socketManager.ts公共文件

import SockJS from 'sockjs-client';
import Stomp from 'stompjs';
import lodash from 'lodash';

function subscribeCallBack(data, subscribes) {
  if (data) {
    let topic = data.headers.destination;
    let funces = subscribes.get(topic);
    funces.forEach((func) => {
      func(data);
    });
  }
}

let clientManager = {
  client: null,
  connecting: false, //是否正在连接
  subscribes: new Map(), //订阅列表
  subscribe(topic, onMessage) {
    if (this.client != null && this.client.connected == true) {
      //已连接状态
      console.log('增加订阅 已连接状态');
      if (!this.subscribes.has(topic)) {
        this.client.subscribe(topic, (data) =>
          subscribeCallBack(data, this.subscribes)
        );
        this.subscribes.set(topic, [onMessage]);
      } else {
        let funces = this.subscribes.get(topic);
        funces.push(onMessage);
      }
    } else {
      //未连接状态
      console.log('增加订阅 未连接状态');
      if (!this.subscribes.has(topic)) {
        this.subscribes.set(topic, [onMessage]);
      } else {
        let funces = this.subscribes.get(topic);
        funces.push(onMessage);
      }
    }
  },
  subscribesAll() {
    console.log('订阅全部');
    if (lodash.isEmpty(this.client) || this.client.connected != true) {
      return;
    }
    let subscribes = this.subscribes;
    for (let topic of subscribes.keys()) {
      this.client.subscribe(topic, (data) =>
        subscribeCallBack(data, subscribes)
      );
    }
  },
  disconnect() {
    console.log('断开连接');
    if (lodash.isEmpty(this.client) || this.client.connected != true) {
      return;
    }
    this.client.disconnect();
    this.subscribes = new Map();
  },
  connect(onSuccess, onDisconnect) {
    try {
      if (this.connecting == true) {
        console.log('正在连接中');
        return;
      }
      this.connecting = true;
      const loginInfo = JSON.parse(sessionStorage.getItem('登录信息'));
      const employeeId = loginInfo && loginInfo.employeeId;
      if (lodash.isEmpty(this.client) || this.client.connected != true) {
        //未连接状态
        let socket = new SockJS(`xxx/endpointSocket`, null, {
          timeout: 6000
        });
        let stompClient = Stomp.over(socket);
        stompClient.debug = null;
        console.log('开始连接');
        stompClient.connect(
          {
            //和后台约定的参数客户id
            Authorization: employeeId
          },
          () => {
            this.client = stompClient;
            console.log('连接成功');
            this.subscribesAll(); //连接成功后开始订阅所有内容
            if (onSuccess != null && onSuccess != undefined) {
              onSuccess();
            }
          },
          (error) => this.errorCallBack(error, onSuccess, onDisconnect)
        );
      } else if (this.client != null && this.client.connected == true) {
        //已连接状态直接调用回调
        onSuccess();
      }
    } catch (err) {
      console.log('连接异常', err);
    } finally {
      this.connecting = false;
    }
  },
  errorCallBack(error, onSuccess, onDisconnect) {
    console.log('连接失败');
    if (onDisconnect != null && onDisconnect != undefined) {
      onDisconnect();
    }
     setTimeout(() => {
       //自动重连
       console.log('重新连接中');
       this.connect(onSuccess, onDisconnect);
     }, 10000);
  }
};

export default clientManager;

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

4、公共文件引入

 import socketmanager from './socketManager';
1

5、直接使用

const openNotification = (type, msg) => {
  notification[type]({
    message: '系统消息',
    description: msg,
    duration: 5
  });
};
componentWillMount() {
	//引入公共文件
    socketmanager.connect(this.onSuccess, this.onDisconnect);
  }
-----------------------------------------------------------------------
onSuccess = () => {
 // console.log('成功');
 const loginInfo = JSON.parse(sessionStorage.getItem('登录信息'));
 const employeeId = loginInfo && loginInfo.employeeId;
 //协商接口参数
 let topic = `/user/${employeeId}/queue`;
 var music = new Audio();
 music.src = newOrder; // 声音文件（随便都行）
 socketmanager.subscribe(topic, (data) => {
   if (data) {
     //do something
     // console.log(data, 'data');
     if (data.body == 'order') {
       let autoplay = true;
       music
         .play()
         .then(() => {
           // 支持自动播放
           autoplay = true;
         })
         .catch((err) => {
           // 不支持自动播放
           autoplay = false;
         })
         .finally(() => {
           music.remove();
           // 告诉调用者结果
         });
       if (!autoplay) {
         message.info('不支持支持语音自动播放');
       }
       openNotification('success', '您有一笔新的订单请注意查收!');
     }
   }
 });
};
onDisconnect = () => {
 console.log('失败');
};
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

参考链接：
基于sockjs的前端WebSocket二次封装

总结

如有问题随时反馈。

工作是为了生活，但生活不是为了工作！慢慢积累，慢慢成长，慢慢蜕变。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/在线问答5/article/detail/770800