内容简介:开门见山,假设一个直播间同时500W人在线,那么1秒钟1000条弹幕,那么弹幕系统的推送频率就是:使用Go做WebSocket开发无非就是三种情况:
(1)业务复杂度介绍
开门见山,假设一个直播间同时500W人在线,那么1秒钟1000条弹幕,那么弹幕系统的推送频率就是: 500W * 1000条/秒=50亿条/秒
,想想B站2019跨年晚会那次弹幕系统得是多么的NB,况且一个大型网站不可能只有一个直播间!
使用Go做WebSocket开发无非就是三种情况:
- 使用Go原生自带的库,也就是
golang.org/x/net
,但是这个官方库真是出了奇Bug多 - 使用GitHub大佬
gorilla/websocket
库,可以结合到某些Web开发框架,比如Gin、iris等,只要使用的框架式基于golang.org/net
的,那么这个库就可以与这个框架结合 - 手撸一个WebSocket框架
根据估算结果,弹幕推送量很大的时候,Linux内核将会出现瓶颈,因为 Linux 内核发送TCP包的时候极限包发送频率是100W。因此可以将同一秒内的弹幕消息合并为1条推送,减少网络小数据包的发送,从而降低推送频率。
弹幕系统需要维护在线的用户长连接来实现定向推送到在线的用户,通常是使用Hash字典结构,通常推送消息就是遍历在线用的Hash字典。在弹幕推送期间用户在不断的上下线,为了维护上线用户,那么就得不断的修改Hash字典,不断地进行锁操作,用户量过大导致锁瓶颈。因此可以将整个Hash结构拆分为多个Hash结构,分别对多个Hash结构加不同的锁,并且使用读写锁替代互斥锁。
通常服务器与客户端交互使用JSON结构,那么需要不断的编码解码JSON数据,这将会导致CPU瓶颈。将消息先进行合并,然后进行编码,最后轮询Hash结构进行推送。
以上是单体架构存在的问题,为了支持更多的用户负载,通常弹幕系统采用分布式架构,进行弹性扩容缩容。
(2)推送还是拉取?
如果是客户端拉取服务器端数据,那么将会存在以下几个问题:
- 直播在线人数多就意味着消息数据更新频率高,拉取消息意味着弹幕无法满足时效性
- 如果很多客户端同时拉取,那么服务器端的压力无异于DDOS
- 一个弹幕系统应该是通用的,因此对于直播间弹幕较少的场景,意味着消息数据拉取请求都是无效的
因此我们考虑推送模式:当数据发生更新的时候服务器端主动推送到客户端,这样可以有效减少客户端的请求次数。如果需要实现消息推送,那么就意味着服务器端维护大量的长连接。
(3)为什么使用WebSocket?
实现弹幕消息的实时更新一定是使用Socket的方式,那么为啥要使用WebSocket呢?现在大部分直播应用的开发都是跨平台的,然而跨平台的开发框架本质就是Web开发,那么一定离不开WebSocket,而且一部分用户会选择在Web端看视频,比如Bilibili,现如今也有一些桌面应用是用Electron等跨平台框架开发的,比如Lark飞书等,因此实现消息推送的最佳方案就是使用WebSocket。
使用WebSocket可以轻松的维持服务器端长连接,其次WebSocket是架构在HTTP协议之上的,并且也可以使用HTTPS方式,因此WebSocket是可靠传输,并且不需要开发者关注底层细节。
为啥要使用Go搞WebSocket呢?首先说到WebSocket你可能会想到Node.js,但是Node.js是单线程模型,如果实现高并发,不得不创建多个Node.js进程,但是这又不容易服务端遍历整个连接集合;如果使用 Java 就会显得比较笨重,Java项目的部署,编写Dockerfile都不如Go的目标二进制更加简洁,并且Go协程很容易实现高并发,上一章说到Go语言目前也有成熟的WebSocket轮子。
(4)服务端基本Demo
首先搭建好一个框架:
package main import ( "fmt" "net/http" ) func main() { fmt.Println("Listen localhost:8080") // 注册一个用于WebSocket的路由,实际业务中不可能只有一个路由 http.HandleFunc("/messages", messageHandler) // 监听8080端口,没有实现服务异常处理器,因此第二个参数是nil http.ListenAndServe("localhost:8080", nil) } func messageHandler(response http.ResponseWriter, request *http.Request) { // TODO: 实现消息处理 response.Write([]byte("HelloWorld")) }
然后完善messageHandler函数:
func messageHandler(response http.ResponseWriter, request *http.Request) { var upgrader = websocket.Upgrader{ // 允许跨域 CheckOrigin: func(resquest *http.Request) bool { return true }, } // 建立连接 conn, err := upgrader.Upgrade(response, request, nil) if err != nil { return } // 收发消息 for { // 读取消息 _, bytes, err := conn.ReadMessage() if err != nil { _ = conn.Close() } // 写入消息 err = conn.WriteMessage(websocket.TextMessage, bytes) if err != nil { _ = conn.Close() } } }
现在基本上实现了WebSocket功能,但是websocket的原生API不是线程安全的(Close方法是线程安全的,并且是可重入的),并且其他模块无法复用业务逻辑,因此进行封装:
- 封装Connection对象描述一个WebSocket连接
- 为Connection对象提供线程安全的关闭、接收、发送API
// main.go package main import ( "bluemiaomiao.github.io/websocket-go/service" "fmt" "net/http" "github.com/gorilla/websocket" ) func main() { fmt.Println("Listen localhost:8080") http.HandleFunc("/messages", messageHandler) _ = http.ListenAndServe("localhost:8080", nil) } func messageHandler(response http.ResponseWriter, request *http.Request) { var upgrader = websocket.Upgrader{ // 允许跨域 CheckOrigin: func(resquest *http.Request) bool { return true }, } // 建立连接 conn, err := upgrader.Upgrade(response, request, nil) wsConn, err := service.Create(conn) if err != nil { return } // 收发消息 for { // 读取消息 msg, err := wsConn.ReadOne() if err != nil { wsConn.Close() } // 写入消息 err = wsConn.WriteOne(msg) if err != nil { _ = conn.Close() } } }
// service/messsage_service.go package service import ( "errors" "github.com/gorilla/websocket" "sync" ) // 封装的连接对象 // // 由于websocket的Close()方法是可重入的,所以可以多次调用,但是关闭Channel的close() // 方法不是可重入的,因此通过isClosed进行判断 // isClosed可能发生资源竞争,因此通过互斥锁避免 // 关闭websocket连接后,也要自动关闭输入输出消息流,因此通过signalCloseLoopChan实现 type Connection struct { conn *websocket.Conn // 具体的连接对象 inputStream chan []byte // 输入流,使用Channel模拟 outputStream chan []byte // 输出流,使用chaneel模拟 signalCloseLoopChan chan byte // 关闭信号 isClosed bool // 是否调用过close()方法 lock sync.Mutex // 简单的锁 } // 用于初始化一个连接对象 func Create(conn *websocket.Conn) (connection *Connection, err error) { connection = &Connection{ conn: conn, inputStream: make(chan []byte, 1000), outputStream: make(chan []byte, 1000), signalCloseLoopChan: make(chan byte, 1), isClosed: false, } // 启动读写循环 go connection.readLoop() go connection.writeLoop() return } // 读取一条消息 func (c *Connection) ReadOne() (msg []byte, err error) { select { case msg = <-(*c).inputStream: case <-(*c).signalCloseLoopChan: err = errors.New("connection is closed") } return } // 写入一条消息 func (c *Connection) WriteOne(msg []byte) (err error) { select { case (*c).outputStream <- msg: case <-(*c).signalCloseLoopChan: err = errors.New("connection is closed") } return } // 关闭连接对象 func (c *Connection) Close() { _ = (*c).conn.Close() (*c).lock.Lock() if !(*c).isClosed { close((*c).signalCloseLoopChan) } (*c).lock.Unlock() } // 读取循环 func (c *Connection) readLoop() { // 不停的读取长连接中的消息,只要存在消息就将其放到队列中 for { _, bytes, err := (*c).conn.ReadMessage() if err != nil { (*c).Close() } select { case <-(*c).signalCloseLoopChan: (*c).Close() case (*c).inputStream <- bytes: } } } // 写入循环 func (c *Connection) writeLoop() { // 只要队列中存在消息,就将其写入 var data []byte for { select { case data = <-(*c).outputStream: case <-(*c).signalCloseLoopChan: (*c).Close() } err := (*c).conn.WriteMessage(websocket.TextMessage, data) if err != nil { _ = (*c).conn.Close() } } }
至此,你已经学会了如何使用Go构建WebSocket服务。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持 码农网
猜你喜欢:- Android开发基于rtmp实现视频直播
- 直播时代,如何自动化保障视频质量?
- 视频技术详解:RTMP H5 直播流技术解析
- 免费!!3天直播外加19个视频,妥妥搞懂ZooKeeper!(限时领)
- 工具 | 字节跳动开源的,直播中台使用的一个视频动画特效 SDK
- 简单直播实现--利用librtmp推音视频流到rtmp服务
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
Large-Scale Inference
Bradley Efron / Cambridge University Press / 2010-8-5 / GBP 48.00
We live in a new age for statistical inference, where modern scientific technology such as microarrays and fMRI machines routinely produce thousands and sometimes millions of parallel data sets, each ......一起来看看 《Large-Scale Inference》 这本书的介绍吧!