Go WebSocket入门+千万级别弹幕系统架构设计

Go实现WebSocket（千万级别弹幕系统架构设计）

1 websocket简介(基于HTTP协议的长连接)

使用WebSocket可以轻松的维持服务器端长连接，其次WebSocket是架构在HTTP协议之上的，并且也可以使用HTTPS方式，因此WebSocket是可靠传输，并且不需要开发者关注底层细节。

websocket具体细节：
①Upgrade：浏览器告知服务器升级为WebSocket协议
②Switch：服务器升级成功后会返回101状态码
③Communicate：浏览器和服务器就可以以WebSocket格式发送数据

还有一种推送数据的方式是SSE：
①SSE(Server Send Event)：服务器单项推送消息，text/event-stream，它是一种流，可以返回多次数据
②使用场景：CI/CD，ChatGPT回答问题

详细文章：推送数据— —WebSocket与SSE

2 弹幕业务的技术选择（推、拉模式）

2.1 客户端拉（服务器压力过大，类似DDoS）

如果是客户端拉取服务器端数据，那么将会存在以下几个问题：

直播在线人数多就意味着消息数据更新频率高，拉取消息意味着弹幕无法满足时效性
如果很多客户端同时拉取，那么服务器端的压力无异于DDOS
一个弹幕系统应该是通用的，因此对于直播间弹幕较少的场景，意味着消息数据拉取请求都是无效的

2.2 服务端推（服务端需要维护大量长连接）

推送模式：当数据发生更新的时候服务器端主动推送到客户端，这样可以有效减少客户端的请求次数。

如果需要实现消息推送，那么就意味着服务器端维护大量的长连接。

3 技术实现（go）

🎆完整代码：

go实现websocket： https://github.com/ziyifast/ziyifast-code_instruction/tree/main/go-demo/go-websocket/1-simple
go实现简易弹幕系统：https://github.com/ziyifast/ziyifast-code_instruction/tree/main/go-demo/go-websocket/2-boardcast

其他教程：Java实现简易聊天室

3.1 前端页面

index.html：

<!DOCTYPE html>
<html>
<head><title>go websocket</title><meta charset="utf-8"/>
</head>
<body>
<script type="text/javascript">var wsUri = "ws://127.0.0.1:7777/ws";var output;function init() {output = document.getElementById("output");testWebSocket();}function testWebSocket() {websocket = new WebSocket(wsUri);websocket.onopen = function (evt) {onOpen(evt)};websocket.onclose = function (evt) {onClose(evt)};websocket.onmessage = function (evt) {onMessage(evt)};websocket.onerror = function (evt) {onError(evt)};}function onOpen(evt) {// writeToScreen("CONNECTED");// doSend("WebSocket rocks");}function onClose(evt) {writeToScreen("DISCONNECTED");}function onMessage(evt) {var message = evt.data;if (message.startsWith("CONNECTED ")) {var connectionId = message.substring("CONNECTED ".length);writeToScreen("CONNECTED: " + connectionId);} else {writeToScreen('<span style="color: blue;">RESPONSE: ' + message + '</span>');}}// function onMessage(evt) {//     writeToScreen('<span style="color: blue;">RESPONSE: '+ evt.data+'</span>');//     // websocket.close();// }function onError(evt) {writeToScreen('<span style="color: red;">ERROR:</span> ' + evt.data);}function doSend(message) {// writeToScreen("SENT: " + message);websocket.send(message);}function writeToScreen(message) {var pre = document.createElement("p");pre.style.wordWrap = "break-word";pre.innerHTML = message;output.appendChild(pre);}window.addEventListener("load", init, false);function sendBtnClick() {var msg = document.getElementById("input").value;doSend(msg);document.getElementById("input").value = '';}function closeBtnClick() {websocket.close();}
</script>
<h2>WebSocket Test</h2>
<input type="text" id="input"></input>
<button onclick="sendBtnClick()">send</button>
<button onclick="closeBtnClick()">close</button>
<div id="output"></div></body>
</html>

3.2 go-websocket实现

1. model/connection.go：封装websocket连接

整体思路：
1. 封装websocket连接为connection
①维护连接的读写channel
②分别启两个协程for循环，一个用于读，一个用于写
//中间多了一层Channel，保证了线程安全
readLoop -> inChannel -> c.ReadMessage拿到data -> c.WriteMessage(data) -> outChannel -> writeLoop从outChannel中拿到data写回同样的数据到对端
2. conn_mgr：实现connection的管理，一旦有消息发送过来，便广播给其他连接，实现弹幕效果

package modelimport ("errors""github.com/google/uuid""github.com/gorilla/websocket""sync"
)/*整体思路：1. 维护连接的读写channel2. 分别启两个协程for循环，一个用于读，一个用于写//中间多了一层Channel，保证了线程安全readLoop -> inChannel -> c.ReadMessage拿到data -> c.WriteMessage(data) -> outChannel -> writeLoop
*/type Connection struct {ConnID stringConn   *websocket.Conn// 读消息队列inChannel chan []byte//写消息队列outChannel chan []byte// 监听Channel是否关闭closeChan chan byte// 标识isClosed boollock     sync.Mutex
}// InitConnection 初始化封装的conn
func InitConnection(conn *websocket.Conn) (c *Connection, err error) {connId, err := uuid.NewUUID()if err != nil {return nil, err}c = &Connection{ConnID:     connId.String(),Conn:       conn,inChannel:  make(chan []byte, 1000),outChannel: make(chan []byte, 1000),closeChan:  make(chan byte),isClosed:   false,}//启动协程读取消息go c.readLoop()go c.writeLoop()return c, nil
}// ReadMessage 读取消息，从inChannel中读取数据（channel保证线程安全，阻塞读取）
func (c *Connection) ReadMessage() (data []byte, err error) {//从inChannel读取数据for {select {case data = <-c.inChannel:return data, nil//监听连接关闭信号，避免一直阻塞读取数据case <-c.closeChan:return nil, errors.New("conn is closed")}}
}// WriteMessage 写消息，将数据写入outChannel（channel保证线程安全,等待write loop从outChannel中获取数据写回连接）
func (c *Connection) WriteMessage(data []byte) (err error) {for {select {case c.outChannel <- data:return nilcase <-c.closeChan:return errors.New("conn is closed")}}
}// 从连接中不断读取数据写入inChannel
func (c *Connection) readLoop() {var (data []byteerr  error)for {if _, data, err = c.Conn.ReadMessage(); err != nil {//读取数据失败，关闭连接c.Close()return}select {//读取到数据写到inChannelcase c.inChannel <- data:case <-c.closeChan:c.Close()}}
}// 从outChannel中不断读取数据并发送数据写回对端
func (c *Connection) writeLoop() {var (data []byteerr  error)for {select {case data = <-c.outChannel:if err = c.Conn.WriteMessage(websocket.TextMessage, data); err != nil {c.Close()return}case <-c.closeChan:c.Close()return}}
}func (c *Connection) Close() {c.Conn.Close()c.lock.Lock()if !c.isClosed {close(c.closeChan)c.isClosed = true}WebSocketMgr.RemoveConnection(c)c.lock.Unlock()
}

2. model/conn_mgr.go

package modelimport ("fmt""sync"
)type connectionMgr struct {connections map[string]*Connectionlock        sync.RWMutex
}var WebSocketMgr = &connectionMgr{connections: make(map[string]*Connection),lock:        sync.RWMutex{},
}func (cm *connectionMgr) AddConnection(conn *Connection) {cm.lock.Lock()defer cm.lock.Unlock()cm.connections[conn.ConnID] = connfmt.Printf("connection %s added\n", conn.ConnID)return
}func (cm *connectionMgr) RemoveConnection(conn *Connection) {cm.lock.Lock()defer cm.lock.Unlock()delete(cm.connections, conn.ConnID)fmt.Printf("connection %s removed\n", conn.ConnID)return
}func (cm *connectionMgr) GetConnection(connID string) (conn *Connection, err error) {cm.lock.RLock()defer cm.lock.RUnlock()conn, ok := cm.connections[connID]if !ok {err = fmt.Errorf("connection not found")return}return
}func (cm *connectionMgr) Boardcast(data []byte) {cm.lock.RLock()defer cm.lock.RUnlock()for _, conn := range cm.connections {if err := conn.WriteMessage(data); err != nil {//if err := conn.WriteMessage([]byte(fmt.Sprintf("[%s] %s", conn.ConnID, string(data)))); err != nil {//TODO 补救或者日志记录，或者忽略return}}
}

3. main.go

package mainimport ("github.com/gorilla/websocket""log""myTest/demo_home/go-demo/go-websocket/2-boardcast/model""net/http"
)var (upgrader = websocket.Upgrader{//允许跨域CheckOrigin: func(r *http.Request) bool {return true}}
)func main() {//模拟简易弹幕系统，注意：为了逻辑简洁，并没有做过多的封装，部分代码设计以及安全监测并不合理http.HandleFunc("/ws", wsHandler)http.ListenAndServe(":7777", nil)
}func wsHandler(w http.ResponseWriter, r *http.Request) {var (conn       *websocket.Connconnection *model.Connectionerr        errordata       []byte)if conn, err = upgrader.Upgrade(w, r, nil); err != nil {return}//初始化连接if connection, err = model.InitConnection(conn); err != nil {return}//注册连接model.WebSocketMgr.AddConnection(connection)// 发送连接ID给前端if err := conn.WriteMessage(websocket.TextMessage, []byte("CONNECTED "+connection.ConnID)); err != nil {log.Println("Error sending connection ID:", err)return}go func(c *model.Connection) {for {if data, err = connection.ReadMessage(); err != nil {return}//广播消息model.WebSocketMgr.Boardcast(data)}}(connection)
}

3.3 效果

1. 启动websocket服务端

2. 分别用chrome、Firefox、edge打开页面，建立websocket连接

CONNECTIONID用于标识不同的websocket长连接

chrome
Firefox
edge

3. 不同浏览器相当于不同用户，chrome用户发起一个弹幕，点击send发送弹幕

在这里插入图片描述

4. 观察其他用户是否接受到弹幕

在这里插入图片描述

自己也会收到自己发的弹幕：

5. edge、firefox用户分别发一个弹幕，观察效果

模拟其他用户发送弹幕

在这里插入图片描述

6. chrome退出直播间，其他用户发送弹幕，它接受不到

chrome用户退出直播间，edge、firefox发送弹幕，chrome用户应该接受不到

在这里插入图片描述
edge、firefox用户发送弹幕：

在这里插入图片描述

后端系统日志：
在这里插入图片描述

4 千万级别弹幕系统设计

4.1 难点(内核、锁、CPU瓶颈)

内核瓶颈：

推送量大：100w在线用户*10条/s = 1000w条/s
linux内核发送TCP的极限包频约为100w条/s

锁瓶颈：

推模式需要维护一个存储了100w条数据的集合，比如map
推送消息遍历整个集合，顺序发送消息，耗时极长
推送期间，客户端仍可以正常上/下线，所以集合需要加锁

CPU瓶颈

浏览器与服务端采用json格式通讯
json编码非常耗cpu
向100w在线用户推送1次，就需要100w次的json encode

4.2 解决方案（多小包合为一个大包）

内核瓶颈：

减少网络小包的发送
将同一秒内的N条消息合并为1条，合并后每秒推送次数只等于在线连接数

锁瓶颈：

将连接分散到多个集合中，每个集合都有自己的锁
多线程并发推送多个集合，避免锁竞争
读写锁取代互斥锁，多个推送任务可以遍历相同集合

cpu瓶颈：

减少重复计算，json编码前置：1次消息编码+100w次推送