自己怎么做視頻收費網(wǎng)站廣告服務(wù)平臺
Go實現(xiàn)WebSocket（千萬級別彈幕系統(tǒng)架構(gòu)設(shè)計）
1 websocket簡介(基于HTTP協(xié)議的長連接)
使用WebSocket可以輕松的維持服務(wù)器端長連接，其次WebSocket是架構(gòu)在HTTP協(xié)議之上的，并且也可以使用HTTPS方式，因此WebSocket是可靠傳輸，并且不需要開發(fā)者關(guān)注底層細(xì)節(jié)。
- websocket具體細(xì)節(jié)：
①Upgrade：瀏覽器告知服務(wù)器升級為WebSocket協(xié)議
②Switch：服務(wù)器升級成功后會返回101狀態(tài)碼
③Communicate：瀏覽器和服務(wù)器就可以以WebSocket格式發(fā)送數(shù)據(jù)
- 還有一種推送數(shù)據(jù)的方式是SSE：
①SSE(Server Send Event)：服務(wù)器單項推送消息，text/event-stream，它是一種流，可以返回多次數(shù)據(jù)
②使用場景：CI/CD，ChatGPT回答問題
詳細(xì)文章：推送數(shù)據(jù)— —WebSocket與SSE
2 彈幕業(yè)務(wù)的技術(shù)選擇（推、拉模式）
2.1 客戶端拉（服務(wù)器壓力過大，類似DDoS）
如果是客戶端拉取服務(wù)器端數(shù)據(jù)，那么將會存在以下幾個問題：
- 直播在線人數(shù)多就意味著消息數(shù)據(jù)更新頻率高，拉取消息意味著彈幕無法滿足時效性
- 如果很多客戶端同時拉取，那么服務(wù)器端的壓力無異于DDOS
- 一個彈幕系統(tǒng)應(yīng)該是通用的，因此對于直播間彈幕較少的場景，意味著消息數(shù)據(jù)拉取請求都是無效的
2.2 服務(wù)端推（服務(wù)端需要維護(hù)大量長連接）
推送模式：當(dāng)數(shù)據(jù)發(fā)生更新的時候服務(wù)器端主動推送到客戶端，這樣可以有效減少客戶端的請求次數(shù)。
- 如果需要實現(xiàn)消息推送，那么就意味著服務(wù)器端維護(hù)大量的長連接。
3 技術(shù)實現(xiàn)（go）
🎆完整代碼：
- go實現(xiàn)websocket：
https://github.com/ziyifast/ziyifast-code_instruction/tree/main/go-demo/go-websocket/1-simple - go實現(xiàn)簡易彈幕系統(tǒng)：
https://github.com/ziyifast/ziyifast-code_instruction/tree/main/go-demo/go-websocket/2-boardcast
其他教程：Java實現(xiàn)簡易聊天室
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實現(xiàn)
1. model/connection.go：封裝websocket連接
整體思路：
1. 封裝websocket連接為connection
①維護(hù)連接的讀寫channel
②分別啟兩個協(xié)程for循環(huán)，一個用于讀，一個用于寫
//中間多了一層Channel，保證了線程安全
readLoop -> inChannel -> c.ReadMessage拿到data -> c.WriteMessage(data) -> outChannel -> writeLoop從outChannel中拿到data寫回同樣的數(shù)據(jù)到對端
2. conn_mgr：實現(xiàn)connection的管理，一旦有消息發(fā)送過來，便廣播給其他連接，實現(xiàn)彈幕效果
package modelimport ("errors""github.com/google/uuid""github.com/gorilla/websocket""sync"
)/*整體思路：1. 維護(hù)連接的讀寫channel2. 分別啟兩個協(xié)程for循環(huán)，一個用于讀，一個用于寫//中間多了一層Channel，保證了線程安全readLoop -> inChannel -> c.ReadMessage拿到data -> c.WriteMessage(data) -> outChannel -> writeLoop
*/type Connection struct {ConnID stringConn *websocket.Conn// 讀消息隊列inChannel chan []byte//寫消息隊列outChannel chan []byte// 監(jiān)聽Channel是否關(guān)閉closeChan chan byte// 標(biāo)識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,}//啟動協(xié)程讀取消息go c.readLoop()go c.writeLoop()return c, nil
}// ReadMessage 讀取消息，從inChannel中讀取數(shù)據(jù)（channel保證線程安全，阻塞讀取）
func (c *Connection) ReadMessage() (data []byte, err error) {//從inChannel讀取數(shù)據(jù)for {select {case data = <-c.inChannel:return data, nil//監(jiān)聽連接關(guān)閉信號，避免一直阻塞讀取數(shù)據(jù)case <-c.closeChan:return nil, errors.New("conn is closed")}}
}// WriteMessage 寫消息，將數(shù)據(jù)寫入outChannel（channel保證線程安全,等待write loop從outChannel中獲取數(shù)據(jù)寫回連接）
func (c *Connection) WriteMessage(data []byte) (err error) {for {select {case c.outChannel <- data:return nilcase <-c.closeChan:return errors.New("conn is closed")}}
}// 從連接中不斷讀取數(shù)據(jù)寫入inChannel
func (c *Connection) readLoop() {var (data []byteerr error)for {if _, data, err = c.Conn.ReadMessage(); err != nil {//讀取數(shù)據(jù)失敗，關(guān)閉連接c.Close()return}select {//讀取到數(shù)據(jù)寫到inChannelcase c.inChannel <- data:case <-c.closeChan:c.Close()}}
}// 從outChannel中不斷讀取數(shù)據(jù)并發(fā)送數(shù)據(jù)寫回對端
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 補(bǔ)救或者日志記錄，或者忽略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() {//模擬簡易彈幕系統(tǒng)，注意：為了邏輯簡潔，并沒有做過多的封裝，部分代碼設(shè)計以及安全監(jiān)測并不合理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)// 發(fā)送連接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服務(wù)端
2. 分別用chrome、Firefox、edge打開頁面，建立websocket連接
CONNECTIONID用于標(biāo)識不同的websocket長連接
- chrome
- Firefox
- edge
3. 不同瀏覽器相當(dāng)于不同用戶，chrome用戶發(fā)起一個彈幕，點擊send發(fā)送彈幕
4. 觀察其他用戶是否接受到彈幕
自己也會收到自己發(fā)的彈幕：
5. edge、firefox用戶分別發(fā)一個彈幕，觀察效果
模擬其他用戶發(fā)送彈幕
6. chrome退出直播間，其他用戶發(fā)送彈幕，它接受不到
chrome用戶退出直播間，edge、firefox發(fā)送彈幕，chrome用戶應(yīng)該接受不到
edge、firefox用戶發(fā)送彈幕：
后端系統(tǒng)日志：
4 千萬級別彈幕系統(tǒng)設(shè)計
4.1 難點(內(nèi)核、鎖、CPU瓶頸)
- 內(nèi)核瓶頸：
- 推送量大：100w在線用戶*10條/s = 1000w條/s
- linux內(nèi)核發(fā)送TCP的極限包頻約為100w條/s
- 鎖瓶頸：
- 推模式需要維護(hù)一個存儲了100w條數(shù)據(jù)的集合，比如map
- 推送消息遍歷整個集合，順序發(fā)送消息，耗時極長
- 推送期間，客戶端仍可以正常上/下線，所以集合需要加鎖
- CPU瓶頸
- 瀏覽器與服務(wù)端采用json格式通訊
- json編碼非常耗cpu
- 向100w在線用戶推送1次，就需要100w次的json encode
4.2 解決方案（多小包合為一個大包）
- 內(nèi)核瓶頸：
- 減少網(wǎng)絡(luò)小包的發(fā)送
- 將同一秒內(nèi)的N條消息合并為1條，合并后每秒推送次數(shù)只等于在線連接數(shù)
- 鎖瓶頸：
- 將連接分散到多個集合中，每個集合都有自己的鎖
- 多線程并發(fā)推送多個集合，避免鎖競爭
- 讀寫鎖取代互斥鎖，多個推送任務(wù)可以遍歷相同集合
- cpu瓶頸：
- 減少重復(fù)計算，json編碼前置：1次消息編碼+100w次推送
4.3 分布式架構(gòu)
如果是單機(jī)架構(gòu)的話：
- 維護(hù)海量的連接必然會耗費很多內(nèi)存
- 消息推送的瞬間也會消耗大量CPU資源
- 消息推送瞬間帶寬可能高達(dá)400-600MB，4-6Gbits（主要瓶頸，即需要萬兆網(wǎng)卡）
因此我們需要分布式架構(gòu)：
- 網(wǎng)關(guān)集群：維護(hù)websocket長連接
- 邏輯集群：基于HTTP/2向gateway網(wǎng)關(guān)集群分發(fā)消息（rpc），與其他服務(wù)的交互等
- 業(yè)務(wù)方
業(yè)務(wù)方->邏輯集群->網(wǎng)關(guān)集群
參考：https://learnku.com/articles/48418