使用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视频直播弹幕内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！