Go实现基于WebSocket的弹幕服务

基于webSocket推送

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

webSocket协议与交互

通讯流程

客户端->upgrade->服务端
客户端<-switching<-服务端
客户端->message->服务端
客户端<-message<-服务端

实现http服务端

1、webSocket是http协议upgrade而来
2、使用http标准库快速实现空接口：/ws

webSocket握手

1、使用webSocket.Upgrader完成协议握手，得到webSocket长连接
2、操作webSocket api，读取客户端消息，然后原样发送回去

封装webSocket

缺乏工程化设计

1、其他代码模块，无法直接操作webSocket连接
2、webSocket连接非线程安全，并发读/写需要同步手段

隐藏细节，封装api

1、封装Connection结构，隐藏webSocket底层连接
2、封装Connection的api，提供Send/Read/Close等线程安全接口

api原理（channel是线程安全的）

1、SendMessage将消息投递到out channel
2、ReadMessage从in channel读取消息

内部原理

1、启动读协程，循环读取webSocket，将消息投递到in channel
2、启动写协程，循环读取out channel，将消息写给webSocket

// server.go
package mainimport (
    "net/http"
    "github.com/gorilla/websocket"
    "./impl"
    "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
        conn *impl.Connection
        data []byte
    )

    //Upgrade:websocket
    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(1 * 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:localhost:7777/ws
    http.HandleFunc("/ws", wsHandler)
    http.ListenAndServe("0.0.0.0:7777", nil)}

// connection.go
package implimport (
    "github.com/gorilla/websocket"
    "sync"
    "github.com/influxdata/platform/kit/errors")var once sync.Oncetype Connection struct {
    wsConn *websocket.Conn
    inChan chan []byte
    outChan chan []byte
    closeChan chan byte
    isClosed bool
    mutex sync.Mutex}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}//APIfunc (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() {
    // 线程安全的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        }

        //阻塞在这里，等待inChan有空位置
        //但是如果writeLoop连接关闭了，这边无法得知
        //conn.inChan <- data

        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        }
        conn.outChan <- data    }
    ERR:
        conn.Close()}