GO实现WebSocket消息推送服务技术分析

2020-01-07 本文已影响0人 Hifun丶

故事背景

在直播行业一个超人气的直播间有时在线人数超过千万，弹幕数量每秒几百万，那么支持这样级别的消息推送技术背后是怎么实现的呢？如果有兴趣可以跟着本文一块分析学习。

ig.png

首先介绍两种获取消息的模式

拉模式（定时轮询访问接口获取数据）

数据更新频率低，则大多数的数据请求时无效的
在线用户数量多，则服务端的查询负载很高
定时轮询拉取，无法满足时效性要求

推模式（向客户端进行数据的推送）

仅在数据更新时，才有推送
需要维护大量的在线长连接
数据更新后，可以立即推送

基于WebSocket协议做推送

浏览器支持的socket编程，轻松维持服务端的长连接
基于TCP协议之上的高层协议，无需开发者关心通讯细节
提供了高度抽象的编程接口，业务开发成本较低

WebSocket协议的交互流程

websocket.png

客户端首先发起一个Http请求到服务端，请求的特殊之处，在于在请求里面带了一个upgrade的字段，告诉服务端，我想生成一个websocket的协议，服务端收到请求后，会给客户端一个握手的确认，返回一个switching，意思允许客户端向websocket协议转换，完成这个协商之后，客户端与服务端之间的底层TCP协议是没有中断的，接下来，客户端可以向服务端发起一个基于websocket协议的消息，服务端也可以主动向客户端发起websocket协议的消息，websocket协议里面通讯的单位就叫message。

传输协议原理

协议升级后，继续复用Http协议的底层socket完成后续通讯
message底层会被切分成多个frame帧进行传输，从协议层面不能传输一个大包，只能切成一个个小包传输
编程时，只需操作message，无需关心frame（属于协议和类库自身去操作的）
框架底层完成TCP网络I/O，WebSocket协议的解析，开发者无需关心

服务端技术选型与考虑

NodeJs

单线程模型（尽管可以多进程），推送性能有限

C/C++

TCP通讯、WebSocket协议实现成本高

Go

多线程，基于协程模型并发
Go语言属于编译型语言，运行速度并不慢
成熟的WebSocket标准库，无需造轮子

基于Go实现WebSocket服务端

用Go语言对WebSocket做一个简单的服务端实现，以及HTML页面进行调试，并对WebSocket封装，这里就直接给出代码了。

WebSocket服务端

/*
* @Author: Hifun
* @Date: 2020/1/6 17:21
 */
package main

import (
    "github.com/golang/impl"
    "github.com/gorilla/websocket"
    "net/http"
    "time"
)

var (
    upgrader = websocket.Upgrader{
        // 允许跨域
        CheckOrigin: func(r *http.Request) bool {
            return true
        },
    }
)

func wsHandler(w http.ResponseWriter, r *http.Request) {
    var (
        wsConn *websocket.Conn
        err    error
        data   []byte
        conn   *impl.Connection
    )
    if wsConn, err = upgrader.Upgrade(w, r, nil); err != nil {
        return
    }

    if conn, err = impl.InitConnection(wsConn); err != nil {
        goto ERR
    }

    go func() {
        var (
            err error
        )
        for {
            if err = conn.WriteMessage([]byte("heartbeat")); err != nil {
                return
            }
            time.Sleep(2 * time.Second)
        }
    }()

    for {
        if data, err = conn.ReadMessage(); err != nil {
            goto ERR
        }
        if err = conn.WriteMessage(data); err != nil {
            goto ERR
        }
    }

ERR:
    conn.Close()
}

func main() {
    http.HandleFunc("/ws", wsHandler)

    http.ListenAndServe(":7777", nil)
}

前端页面 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) { 
            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>

封装WebSocket

这里解释一下为什么要封装起来

1.缺乏工程化设计，其他代码模块无法直接操作Websocket连接。
2.websocket连接非线程安全，并发读/写需要同步手段。(ReadMessage、WriteMessage一次只能有一个在执行)

这里再解释一下线程安全

线程安全: 指多个线程在执行同一段代码的时候采用加锁机制，使每次的执行结果和单线程执行的结果都是一样的，不存在执行程序时出现意外结果。

下面是封装的代码，我这里定义了一个包，放到了GOPATH里

/*
* @Author: Hifun
* @Date: 2020/1/7 16:01
 */
package impl

import (
    "errors"
    "github.com/gorilla/websocket"
    "sync"
)

type Connection struct {
    wsConn    *websocket.Conn
    inChan    chan []byte
    outChan   chan []byte
    closeChan chan byte

    mutex    sync.Mutex
    isClosed bool
}

func InitConnection(wsConn *websocket.Conn) (conn *Connection, err error) {
    conn = &Connection{
        wsConn:    wsConn,
        inChan:    make(chan []byte, 1000),
        outChan:   make(chan []byte, 1000),
        closeChan: make(chan byte, 1),
    }
    // 启动读协程
    go conn.readLoop()
    // 启动写协程
    go conn.writeLoop()
    return
}

// API
func (conn *Connection) ReadMessage() (data []byte, err error) {
    select {
    case data = <-conn.inChan:
    case <-conn.closeChan:
        err = errors.New("Connection is closed")
    }
    return
}
func (conn *Connection) WriteMessage(data []byte) (err error) {
    select {
    case conn.outChan <- data:
    case <-conn.closeChan:
        err = errors.New("Connection is closed")
    }
    return
}

func (conn *Connection) Close() {
    conn.wsConn.Close()
    // 这里只能执行一次
    conn.mutex.Lock()
    if !conn.isClosed {
        close(conn.closeChan)
        conn.isClosed = true
    }
    conn.mutex.Unlock()

}

func (conn *Connection) readLoop() {
    var (
        data []byte
        err  error
    )
    for {
        if _, data, err = conn.wsConn.ReadMessage(); err != nil {
            goto ERR
        }
        // 这里可能阻塞
        select {
        case conn.inChan <- data:
        case <-conn.closeChan:
            // closeChan 关闭进入
            goto ERR
        }

    }
ERR:
    conn.Close()
}

func (conn *Connection) writeLoop() {
    var (
        data []byte
        err  error
    )
    for {
        select {
        case data = <-conn.outChan:
        case <-conn.closeChan:
            goto ERR
        }
        if err = conn.wsConn.WriteMessage(websocket.TextMessage, data); err != nil {
            goto ERR
        }
    }
ERR:
    conn.Close()
}

千万级弹幕系统的架构设计分析

技术难点

内核瓶颈

推送量大：100W在线 * 10条/每秒 = 1000W条/秒
内核瓶颈：linux内核发送TCP的极限包频 ≈ 100W/秒

锁瓶颈

需要维护在线用户集合（100W用户在线），通常是一个字典结构
推送消息即遍历整个集合，顺序发送消息，耗时极长
推送期间，客户端仍旧正常的上下线，集合面临不停的修改，修改需要遍历，所以集合需要上锁

CPU瓶颈

浏览器与服务端之间一般采用的是Json格式去通讯
Json编码非常耗费CPU资源
向100W在线推送一次，则需100W次Json Encode

优化方案

内核瓶颈

减少网络小包的发送，我们将网络上几百字节定义成网络的小包了，小包的问题是对内核和网络的中间设备造成处理的压力。方案是将一秒内N条消息合并成1条消息，合并后，每秒推送数等于在线连接数。

锁瓶颈

大锁拆小锁，将长连接打散到多个集合中去，每个集合都有自己的锁，多线程并发推送集合，线程之间推送的集合不同，所以没有锁的竞争关系，避免锁竞争。
读写锁取代互斥锁，多个推送任务可以并发遍历相同集合

CPU瓶颈

减少重复计算，Json编码前置，1次消息编码+100W次推送，消息合并前置，N条消息合并后，只需要编码一次。

单机架构

simple.png

最外层是在线的长连接，连接到服务端后，打散到多个集合里面存储，我们要发送的消息呢，通过打包后，经过json编码，被多个线程或协程分发到多个集合中去，最终推给了所有的在线连接。

单机瓶颈

维护海量长连接，会花费不少内存
消息推送的瞬时，消耗大量的CPU
消息推送的瞬时带宽高达400-600Mb（4-6Gbits），需要用到万兆网卡，是主要瓶颈

集群

部署多个节点，通过负载均衡，把连接打散到多个服务器上，但推送消息的时候，不知道哪个直播间在哪个节点上，最常用的方式是将消息广播给所有的网关节点，此时就需要做一个逻辑集群。

逻辑集群

基于Http2协议向gateway集群分发消息（Http2支持连接复用，用作RPC性能更佳，即在单个连接上可以做高吞吐的请求应答处理）
基于Http1协议对外提供推送API（Http1更加普及，对业务方更加友好）
整体分布式架构图如下：

http.png

任何业务方通过Http接口调用到逻辑集群，逻辑集群把消息广播给所有网关，各个网关各自将消息推送给在线的连接即可。

本文讲解了开发消息推送服务的难点与解决方案的大体思路，按照整个理论流程下来，基本能实现一套弹幕消息推送的服务。理论远比不上实践，动手敲一遍吧！

GO实现WebSocket消息推送服务技术分析

故事背景

首先介绍两种获取消息的模式

拉模式（定时轮询访问接口获取数据）

推模式（向客户端进行数据的推送）

基于WebSocket协议做推送

WebSocket协议的交互流程

传输协议原理

服务端技术选型与考虑

NodeJs

C/C++

Go

基于Go实现WebSocket服务端

WebSocket服务端

前端页面 html (示例) 测试用

封装WebSocket

这里解释一下为什么要封装起来

这里再解释一下线程安全

下面是封装的代码，我这里定义了一个包，放到了GOPATH里

千万级弹幕系统的架构设计分析

技术难点

内核瓶颈

锁瓶颈

CPU瓶颈

优化方案

内核瓶颈

锁瓶颈

CPU瓶颈

单机架构

单机瓶颈

集群

逻辑集群

猜你喜欢

热点阅读