#Go WebSocket入门+千万级别弹幕系统架构设计
Go实现WebSocket（千万级别弹幕系统架构设计）
1 websocket简介(基于HTTP协议的长连接)
使用WebSocket可以轻松的维持服务器端长连接，其次WebSocket是架构在HTTP协议之上的，并且也可以使用HTTPS方式，因此WebSocket是可靠传输，并且不需要开发者关注底层细节。
详细文章：
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-simple1. go实现简易弹幕系统：https://github.com/ziyifast/ziyifast-code_instruction/tree/main/go-demo/go-websocket/2-boardcast
其他教程：
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 model
import (
"errors"
"github.com/google/uuid"
"github.com/gorilla/websocket"
"sync"
)
/*
整体思路：
1. 维护连接的读写channel
2. 分别启两个协程for循环，一个用于读，一个用于写
//中间多了一层Channel，保证了线程安全
readLoop -> inChannel -> c.ReadMessage拿到data -> c.WriteMessage(data) -> outChannel -> writeLoop
*/
type Connection struct {<!-- -->
ConnID string
Conn *websocket.Conn
// 读消息队列
inChannel chan []byte
//写消息队列
outChannel chan []byte
// 监听Channel是否关闭
closeChan chan byte
// 标识
isClosed bool
lock 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 nil
case <-c.closeChan:
return errors.New("conn is closed")
}
}
}
// 从连接中不断读取数据写入inChannel
func (c *Connection) readLoop() {<!-- -->
var (
data []byte
err error
)
for {<!-- -->
if _, data, err = c.Conn.ReadMessage(); err != nil {<!-- -->
//读取数据失败，关闭连接
c.Close()
return
}
select {<!-- -->
//读取到数据写到inChannel
case c.inChannel <- data:
case <-c.closeChan:
c.Close()
}
}
}
// 从outChannel中不断读取数据并发送数据写回对端
func (c *Connection) writeLoop() {<!-- -->
var (
data []byte
err 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 model
import (
"fmt"
"sync"
)
type connectionMgr struct {<!-- -->
connections map[string]*Connection
lock 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] = conn
fmt.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 main
import (
"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.Conn
connection *model.Connection
err error
data []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长连接
<img src="https://i-blog.csdnimg.cn/blog_migrate/8db194b2ebb620ab52fcb4c89170322e.png" alt="在这里插入图片描述">
4. 观察其他用户是否接受到弹幕
<img src="https://i-blog.csdnimg.cn/blog_migrate/46c6500b06d31a7535f27c7e5239e1ef.png" alt="在这里插入图片描述"> <img src="https://i-blog.csdnimg.cn/blog_migrate/3ddc295aa8201fa7990bd0d2705e5b58.png" alt="在这里插入图片描述"> 自己也会收到自己发的弹幕： <img src="https://i-blog.csdnimg.cn/blog_migrate/4a3ca530b1927c72264f857deeeba35a.png" alt="在这里插入图片描述">
5. edge、firefox用户分别发一个弹幕，观察效果
模拟其他用户发送弹幕
<img src="https://i-blog.csdnimg.cn/blog_migrate/58514cc9535ee83010e7c264acff0614.png" alt="在这里插入图片描述"> <img src="https://i-blog.csdnimg.cn/blog_migrate/f9a939745460736c33686e196b7484f9.png" alt="在这里插入图片描述">
<img src="https://i-blog.csdnimg.cn/blog_migrate/c460d102701d184b1f260f3f7b756b22.png" alt="在这里插入图片描述">
6. chrome退出直播间，其他用户发送弹幕，它接受不到
chrome用户退出直播间，edge、firefox发送弹幕，chrome用户应该接受不到
<img src="https://i-blog.csdnimg.cn/blog_migrate/f65ffac5eb893f00d0f13841381930e7.png" alt="在这里插入图片描述"> edge、firefox用户发送弹幕： <img src="https://i-blog.csdnimg.cn/blog_migrate/dc81d88eb2d2b549e8453458a3f09720.png" alt="在这里插入图片描述">
<img src="https://i-blog.csdnimg.cn/blog_migrate/9799b030da939b697fb91b451fc3c99b.png" alt="在这里插入图片描述"> <img src="https://i-blog.csdnimg.cn/blog_migrate/ac22a866529abf6c7e148693579c62be.png" alt="在这里插入图片描述">
后端系统日志： <img src="https://i-blog.csdnimg.cn/blog_migrate/c0f45edfd4c46b3fb6deba032da7316d.png" alt="在这里插入图片描述">
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次推送
4.3 分布式架构
如果是单机架构的话：
- 维护海量的连接必然会耗费很多内存- 消息推送的瞬间也会消耗大量CPU资源- 消息推送瞬间带宽可能高达400-600MB，4-6Gbits（主要瓶颈，即需要万兆网卡）
因此我们需要分布式架构：
- 网关集群：维护websocket长连接- 逻辑集群：基于HTTP/2向gateway网关集群分发消息（rpc），与其他服务的交互等- 业务方
业务方->逻辑集群->网关集群
参考：https://learnku.com/articles/48418
