Go WebSocket 的使用

WebSocket在 HTML5 游戲和網(wǎng)頁消息推送都使用比較多。WebSocket 是 HTML5 的重要特性，它實(shí)現(xiàn)了基于瀏覽器的遠(yuǎn)程socket，它使瀏覽器和服務(wù)器可以進(jìn)行全雙工通信。
WebSocket 具體的特性和 http 的區(qū)別這里就不多說，可以去自己查一下。

Go 官方?jīng)]有提供對 WebSocket 的支持，必須選擇第三方提供的包?！禛o Web 編程》一書中的例子使用了 golang.org/x/net 下的 websocket 包。 另外一個使用比較多的是 gorilla/websocket ，我接觸的項(xiàng)目是使用的這個。下面我就以 gorilla/websocket 來寫一個簡單的通信示例。

gorilla/websocket 的資料參考：
GitHub：https://github.com/gorilla/websocket
Doc：https://godoc.org/github.com/gorilla/websocket

gorilla/websocket 簡述

Upgrader

Upgrader 用于升級 http 請求，把 http 請求升級為長連接的 WebSocket。結(jié)構(gòu)如下：

type Upgrader struct {
    // 指定升級 websocket 握手完成的超時時間
    HandshakeTimeout time.Duration

    // 指定 io 操作的緩存大小，如果不指定就會自動分配。
    ReadBufferSize, WriteBufferSize int

    // 寫數(shù)據(jù)操作的緩存池，如果沒有設(shè)置值，write buffers 將會分配到鏈接生命周期里。
    WriteBufferPool BufferPool

    //按順序指定服務(wù)支持的協(xié)議，如值存在，則服務(wù)會從第一個開始匹配客戶端的協(xié)議。
    Subprotocols []string

    // 指定 http 的錯誤響應(yīng)函數(shù)，如果沒有設(shè)置 Error 則，會生成 http.Error 的錯誤響應(yīng)。
    Error func(w http.ResponseWriter, r *http.Request, status int, reason error)

    // 請求檢查函數(shù)，用于統(tǒng)一的鏈接檢查，以防止跨站點(diǎn)請求偽造。如果不檢查，就設(shè)置一個返回值為true的函數(shù)。
    // 如果請求Origin標(biāo)頭可以接受，CheckOrigin將返回true。 如果CheckOrigin為nil，則使用安全默認(rèn)值：如果Origin請求頭存在且原始主機(jī)不等于請求主機(jī)頭，則返回false
    CheckOrigin func(r *http.Request) bool

    // EnableCompression 指定服務(wù)器是否應(yīng)嘗試協(xié)商每個郵件壓縮（RFC 7692）。 
    // 將此值設(shè)置為true并不能保證將支持壓縮。 
    // 目前僅支持“無上下文接管”模式
    EnableCompression bool
}

func (*Upgrader) Upgrade

Upgrade 函數(shù)將 http 升級到 WebSocket 協(xié)議。定義如下：

// responseHeader包含在對客戶端升級請求的響應(yīng)中。 
// 使用responseHeader指定cookie（Set-Cookie）和應(yīng)用程序協(xié)商的子協(xié)議（Sec-WebSocket-Protocol）。
// 如果升級失敗，則升級將使用HTTP錯誤響應(yīng)回復(fù)客戶端
// 返回一個 Conn 指針，拿到他后，可使用 Conn 讀寫數(shù)據(jù)與客戶端通信。
func (u *Upgrader) Upgrade(w http.ResponseWriter, r *http.Request, responseHeader http.Header) (*Conn, error)

使用實(shí)例

type WsServer struct {
    ......
    // 定義一個 upgrade 類型用于升級 http 為 websocket
    upgrade  *websocket.Upgrader
}

func NewWsServer() *WsServer {
    ws.upgrade = &websocket.Upgrader{
        ReadBufferSize:  4096,//指定讀緩存區(qū)大小
        WriteBufferSize: 1024,// 指定寫緩存區(qū)大小
        // 檢測請求來源
        CheckOrigin: func(r *http.Request) bool {
            if r.Method != "GET" {
                fmt.Println("method is not GET")
                return false
            }
            if r.URL.Path != "/ws" {
                fmt.Println("path error")
                return false
            }
            return true
        },
    }
    return ws
}

func (self *WsServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    ......
    // 收到 http 請求后 升級 協(xié)議
    conn, err := self.upgrade.Upgrade(w, r, nil)
    if err != nil {
        fmt.Println("websocket error:", err)
        return
    }
    fmt.Println("client connect :", conn.RemoteAddr())
    go self.connHandle(conn)

}

http 服務(wù)

啟動一個 http 服務(wù)有多種方法

  • 第一種
func (self *WsServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    if r.URL.Path != "/v1" {
        httpCode := http.StatusInternalServerError
        reasePhrase := http.StatusText(httpCode)
        fmt.Println("path error ", reasePhrase)
        http.Error(w, reasePhrase, httpCode)
        return
    }
}

func (w *WsServer) Start() (err error) {
    // 使用 net.Listen 監(jiān)聽端口服務(wù)
    w.listener, err = net.Listen("tcp", w.addr)
    if err != nil {
        fmt.Println("net listen error:", err)
        return
    }
    // 啟動 http 服務(wù)， w 參數(shù)類型需要 實(shí)現(xiàn)  Handler 接口，也就是 ServeHTTP 函數(shù)
    err = http.Serve(w.listener, w)
    if err != nil {
        fmt.Println("http serve error:", err)
        return
    }
    return nil
}
  • 第二種
import (
    "fmt"
    "io"
    "net/http"
)
func serveHandle(w http.ResponseWriter, r *http.Request) {
    buf := make([]byte, 1024)
    r.Body.Read(buf)
    fmt.Println("request body", string(buf))
    io.WriteString(w, "hello http server 1")
}
func main() {
    http.HandleFunc("/v1", serveHandle)
    // 直接使用 http 包的 ListenAndServe 函數(shù)監(jiān)聽服務(wù)
    http.ListenAndServe(s.addr, nil)
}

WebSocket 完整代碼

http 服務(wù) + Upgrade 實(shí)現(xiàn) WebSocket

package main
import (
    "fmt"
    "net"
    "net/http"
    "time"
    "github.com/gorilla/websocket"
)
type WsServer struct {
    listener net.Listener
    addr     string
    upgrade  *websocket.Upgrader
}
func NewWsServer() *WsServer {
    ws := new(WsServer)
    ws.addr = "0.0.0.0:10215"
    ws.upgrade = &websocket.Upgrader{
        ReadBufferSize:  4096,
        WriteBufferSize: 1024,
        CheckOrigin: func(r *http.Request) bool {
            if r.Method != "GET" {
                fmt.Println("method is not GET")
                return false
            }
            if r.URL.Path != "/ws" {
                fmt.Println("path error")
                return false
            }
            return true
        },
    }
    return ws
}

func (self *WsServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    if r.URL.Path != "/ws" {
        httpCode := http.StatusInternalServerError
        reasePhrase := http.StatusText(httpCode)
        fmt.Println("path error ", reasePhrase)
        http.Error(w, reasePhrase, httpCode)
        return
    }
    conn, err := self.upgrade.Upgrade(w, r, nil)
    if err != nil {
        fmt.Println("websocket error:", err)
        return
    }
    fmt.Println("client connect :", conn.RemoteAddr())
    go self.connHandle(conn)

}
func (self *WsServer) connHandle(conn *websocket.Conn) {
    defer func() {
        conn.Close()
    }()
    stopCh := make(chan int)
    go self.send(conn, stopCh)
    for {
        conn.SetReadDeadline(time.Now().Add(time.Millisecond * time.Duration(5000)))
        _, msg, err := conn.ReadMessage()
        if err != nil {
            close(stopCh)
            // 判斷是不是超時
            if netErr, ok := err.(net.Error); ok {
                if netErr.Timeout() {
                    fmt.Printf("ReadMessage timeout remote: %v\n", conn.RemoteAddr())
                    return
                }
            }
            // 其他錯誤，如果是 1001 和 1000 就不打印日志
            if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
                fmt.Printf("ReadMessage other remote:%v error: %v \n", conn.RemoteAddr(), err)
            }
            return
        }
        fmt.Println("收到消息：", string(msg))
    }
}

//測試一次性發(fā)送 10萬條數(shù)據(jù)給 client, 如果不使用 time.Sleep browser 過了超時時間會斷開
func (self *WsServer) send10(conn *websocket.Conn) {
    for i := 0; i < 100000; i++ {
        data := fmt.Sprintf("hello websocket test from server %v", time.Now().UnixNano())
        err := conn.WriteMessage(1, []byte(data))
        if err != nil {
            fmt.Println("send msg faild ", err)
            return
        }
        // time.Sleep(time.Millisecond * 1)
    }
}

func (self *WsServer) send(conn *websocket.Conn, stopCh chan int) {
    self.send10(conn)
    for {
        select {
        case <-stopCh:
            fmt.Println("connect closed")
            return
        case <-time.After(time.Second * 1):
            data := fmt.Sprintf("hello websocket test from server %v", time.Now().UnixNano())
            err := conn.WriteMessage(1, []byte(data))
            fmt.Println("sending....")
            if err != nil {
                fmt.Println("send msg faild ", err)
                return
            }
        }
    }
}

func (w *WsServer) Start() (err error) {
    w.listener, err = net.Listen("tcp", w.addr)
    if err != nil {
        fmt.Println("net listen error:", err)
        return
    }
    err = http.Serve(w.listener, w)
    if err != nil {
        fmt.Println("http serve error:", err)
        return
    }
    return nil
}

func main() {
    ws := NewWsServer()
    ws.Start()
}

WebSocket 客戶端

  • 更新中
    VILEngine.js 文件
let VIL = (function () {
    let VIL = {
    };

    function DefaultWebSocket(host, call) {
        let _host = host;
        let _isOpen = false;
        let _bufQueue = [];
        let _bufCap = 100;
        let _call = null;
        if("undefined" !== typeof call && call !== null){
            _call = call
        }else{
            _call = {
                onConnect:function (e) {
                    console.log("connect success ", e);
                },
                onDisconnect:function (e) {
                    console.log("disconnect ", e);
                },
                onMsg:function (data) {
                    //console.log("receive message ", data)
                }
            }
        }


        let _socket = new WebSocket(_host);
        _socket.binaryType = "arraybuffer";
        /**
         * 設(shè)置發(fā)送消息緩存隊(duì)列的容量
         * @param {number} cap
         * @constructor
         */
        this.setBufferCap = function(cap){
            if("number" !== typeof cap ){
                console.error("parameter type is not number ");
                return ;
            }
            if(cap < 0){
                console.error("parameter value can not less then 0");
                return ;
            }
            _bufCap = cap;
        };

        /**
         * 發(fā)送消息
         * @param {string | ArrayBuffer } data
         * @constructor
         */
        this.send = function(data){
            if(_isOpen && _socket){
                _socket.send("");
            }else{
                if (_bufQueue < _bufCap){
                    _bufQueue.push(data);
                }
            }
        };

        this.close = function(){
            _socket.close(1000, "normal");
        };

        _socket.onopen = function(even){
            _isOpen = true;
            _call.onConnect(even);
            while (_bufQueue > 0){
                _socket.send(_bufQueue.shift());
            }
        };

        _socket.onmessage = function(e){
            let data = e.data;
            _call.onMsg(data);
        };

        /**
         * 收到關(guān)閉連接
         * @param even
         */
        _socket.onclose = function(even){
            _isOpen = false;
            _call.onDisconnect({host:_host, event:even});
        };

        /**
         * 收到錯誤
         * @param err
         */
        _socket.onerror = function(err){
            _isOpen = false;
            _call.onDisconnect({host:_host, event:err});
        };
    }

    try{
        VIL.EngineSocket = DefaultWebSocket ;
    }catch (e) {
        console.error("VILEngine error ", e);
    }

    return VIL;
})();

websocket.html 文件

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
</body>
<script type="text/javascript" src="VILEngine.js"></script>
<script>
    let counter  = 0;
    let isConnect = false;
    let handler = {
        onConnect:function (e) {
            isConnect = true;
            console.log("handler connect success ", e);
            var se = setInterval(function () {
                if(isConnect === false){
                    clearInterval(se);
                }
                console.log("setInterval", Date.now());
                socket.send("web browser setInterval");
            }, 3000)
        },
        onDisconnect:function (e) {
            isConnect = false;
            console.log("handler disconnect ", e);
        },
        onMsg:function (data) {
            counter++;
            if(counter >= 2000){
                counter = 0;
                console.log("handler receive message ", data)
            }
        }
    };
    let socket = new VIL.EngineSocket("ws://127.0.0.1:10215/ws", handler);
</script>
</html>

WebSocket 服務(wù)端 和 客戶端在 GitHub 上有源碼，需要的可以訪問
https://github.com/vilsongwei/practiceDemo/tree/master/websockTest

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容