如何使用Go基于WebSocket构建视频直播弹幕系统-创新互联
创新互联www.cdcxhl.cn八线动态BGP香港云服务器提供商，新人活动买多久送多久，划算不套路！
本篇文章给大家分享的是有关如何使用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服务。
以上就是如何使用Go基于WebSocket构建视频直播弹幕系统，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注创新互联-成都网站建设公司行业资讯频道。
本文题目：如何使用Go基于WebSocket构建视频直播弹幕系统-创新互联
链接分享：http://shouzuofang.com/article/dddjec.html