使用Go基于WebSocket构建千万级视频直播弹幕系统的代码详解

 更新时间:2020-08-17 13:15:16   作者:佚名   我要评论(0)

(1)业务复杂度介绍
开门见山,假设一个直播间同时500W人在线,那么1秒钟1000条弹幕,那么弹幕系统的推送频率就是: 500W * 1000条/秒=50亿条/秒 ,想想B站2019跨年

(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视频直播弹幕内容请搜索脚本之家以前的文章或继续浏览下面的果博东方相关的文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:

  • Websocket直播间聊天室教程 GoEasy快速实现聊天室
  • 利用 Go 语言编写一个简单的 WebSocket 推送服务
  • 利用Go语言搭建WebSocket服务端方法示例
  • go的websocket实现原理与用法详解
  • golang基于websocket实现的简易聊天室程序

果博东方相关的文章

  • 使用Go基于WebSocket构建千万级视频直播弹幕系统的代码详解

    使用Go基于WebSocket构建千万级视频直播弹幕系统的代码详解

    (1)业务复杂度介绍 开门见山,假设一个直播间同时500W人在线,那么1秒钟1000条弹幕,那么弹幕系统的推送频率就是: 500W * 1000条/秒=50亿条/秒 ,想想B站2019跨年
    2020-08-17
  • Go 语言中关于接口的三个

    Go 语言中关于接口的三个

    我的在线博客:http://golang.iswbm.com 我的 Github:github.com/iswbm/GolangCodingTime 1. 对方法的调用限制 接口是一组固定的方法集,由于静态类型的限制,接
    2020-08-17
  • Go 语言json.Unmarshal 遇到的小问题(推荐)

    Go 语言json.Unmarshal 遇到的小问题(推荐)

    1.问题现象描述 使用 json.Unmarshal(),反序列化时,出现了科学计数法,参考代码如下: jsonStr := `{"number":1234567}` result := make(map[string]interface{
    2020-08-17
  • 深度解密 Go 语言中的 sync.map

    深度解密 Go 语言中的 sync.map

    工作中,经常会碰到并发读写 map 而造成 panic 的情况,为什么在并发读写的时候,会 panic 呢?因为在并发读写的情况下,map 里的数据会被写乱,之后就是 Garbage i
    2020-08-17
  • Xshell与Xftp免费下载安装及步骤(图文)

    Xshell与Xftp免费下载安装及步骤(图文)

    1.进入Xshell的官网:http://www.netsarang.com/zh/ 2.选择你需要的软件进行下载如:Xshell 3.选择家庭和学校用户的免费许可证进行免费授权 https:/
    2020-08-17
  • 详解go 动态数组 二维动态数组

    详解go 动态数组 二维动态数组

    go使用动态数组还有点麻烦,比python麻烦一点,需要先定义。 动态数组申明 var dynaArr []string 动态数组添加成员 dynaArr = append(dynaArr, "one") ```go
    2020-08-17
  • golang中使用sync.Map的方法

    golang中使用sync.Map的方法

    背景 go中map数据结构不是线程安全的,即多个goroutine同时操作一个map,则会报错,因此go1.9之后诞生了sync.Map sync.Map思路来自java的ConcurrentHashMap 接口 sy
    2020-08-17
  • Go语言实现基于websocket浏览器通知功能

    Go语言实现基于websocket浏览器通知功能

    当我们使用了websocket技术实现了实时的推送访客来到消息,但是我们只能随时看着页面不离开才能知道有访客来到 现在增加浏览器的通知消息,就可以只要打开着界面,不用
    2020-08-17
  • 详解Golang开启http服务的三种方式

    详解Golang开启http服务的三种方式

    前言 都说go标准库实用,Api设计简洁。这次就用go 标准库中的net/http包实现一个简洁的http web服务器,包括三种版本。 v1最简单版 直接使用http.HandleFunc(parter
    2020-08-17
  • GO语言实现标题闪烁效果

    GO语言实现标题闪烁效果

    在实现客服系统的过程中,需要有新消息的时候标题栏闪烁提示 因为聊天框是被iframe框进去的页面,所以在聊天框中收到websocket消息以后要把消息发送给父集页面,在父页
    2020-08-17

最新评论

RNG战胜OMG玄彬孙艺珍同框nba选秀于汉超英语APP葫芦兄弟邮票