基於websocket單臺機器支持百萬鏈接分佈式聊天(IM)系統

時間  2019-11-07
標籤 基於 websocket 機器 支持 百萬 鏈接 分佈式 聊天 系統 欄目 系統架構 简体版
原文   原文鏈接

基於websocket單臺機器支持百萬鏈接分佈式聊天(IM)系統

本文將介紹如何實現一個基於websocket分佈式聊天(IM)系統。php

使用golang實現websocket通信，單機能夠支持百萬鏈接，使用gin框架、nginx負載、能夠水平部署、程序內部相互通信、使用grpc通信協議。html

本文內容比較長，若是直接想clone項目體驗直接進入項目體驗 goWebSocket項目下載 ,文本從介紹webSocket是什麼開始，而後開始介紹這個項目，以及在Nginx中配置域名作webSocket的轉發，而後介紹如何搭建一個分佈式系統。前端

目錄

一、項目說明

1.1 goWebSocket

本文將介紹如何實現一個基於websocket聊天(IM)分佈式系統。

使用golang實現websocket通信，單機支持百萬鏈接，使用gin框架、nginx負載、能夠水平部署、程序內部相互通信、使用grpc通信協議。

  • 通常項目中webSocket使用的架構圖

網站架構圖

1.2 項目體驗

二、介紹webSocket

2.1 webSocket 是什麼

WebSocket 協議在2008年誕生，2011年成爲國際標準。全部瀏覽器都已經支持了。

它的最大特色就是，服務器能夠主動向客戶端推送信息，客戶端也能夠主動向服務器發送信息，是真正的雙向平等對話，屬於服務器推送技術的一種。

  • HTTP和WebSocket在通信過程的比較

HTTP協議和WebSocket比較

  • HTTP和webSocket都支持配置證書，ws:// 無證書 wss:// 配置證書的協議標識

HTTP協議和WebSocket比較

2.2 webSocket的兼容性

  • 瀏覽器的兼容性，開始支持webSocket的版本

瀏覽器開始支持webSocket的版本

  • 服務端的支持

golang、java、php、node.js、python、nginx 都有不錯的支持

  • Android和IOS的支持

Android可使用java-webSocket對webSocket支持

iOS 4.2及更高版本具備WebSockets支持

2.3 爲何要用webSocket

    1. 從業務上出發，須要一個主動通達客戶端的能力
目前大多數的請求都是使用HTTP，都是由客戶端發起一個請求，有服務端處理，而後返回結果，不能夠服務端主動向某一個客戶端主動發送數據

服務端處理一個請求

    1. 大多數場景咱們須要主動通知用戶，如:聊天系統、用戶完成任務主動告訴用戶、一些運營活動須要通知到在線的用戶
    1. 能夠獲取用戶在線狀態
    1. 在沒有長連接的時候經過客戶端主動輪詢獲取數據
    1. 能夠經過一種方式實現，多種不一樣平臺(H5/Android/IOS)去使用

2.4 webSocket創建過程

    1. 客戶端先發起升級協議的請求

客戶端發起升級協議的請求，採用標準的HTTP報文格式，在報文中添加頭部信息

Connection: Upgrade代表鏈接須要升級

Upgrade: websocket須要升級到 websocket協議

Sec-WebSocket-Version: 13 協議的版本爲13

Sec-WebSocket-Key: I6qjdEaqYljv3+9x+GrhqA== 這個是base64 encode 的值，是瀏覽器隨機生成的，與服務器響應的 Sec-WebSocket-Accept對應

# Request Headers
Connection: Upgrade
Host: im.91vh.com
Origin: http://im.91vh.com
Pragma: no-cache
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits
Sec-WebSocket-Key: I6qjdEaqYljv3+9x+GrhqA==
Sec-WebSocket-Version: 13
Upgrade: websocket

瀏覽器 Network

    1. 服務器響應升級協議

服務端接收到升級協議的請求，若是服務端支持升級協議會作以下響應

返回:

Status Code: 101 Switching Protocols 表示支持切換協議

# Response Headers
Connection: upgrade
Date: Fri, 09 Aug 2019 07:36:59 GMT
Sec-WebSocket-Accept: mB5emvxi2jwTUhDdlRtADuBax9E=
Server: nginx/1.12.1
Upgrade: websocket
    1. 升級協議完成之後，客戶端和服務器就能夠相互發送數據

websocket接收和發送數據

三、如何實現基於webSocket的長連接系統

3.1 使用go實現webSocket服務端

3.1.1 啓動端口監聽

  • websocket須要監聽端口，因此須要在golang 成功的 main 函數中用協程的方式去啓動程序
  • main.go 實現啓動
go websocket.StartWebSocket()
  • init_acc.go 啓動程序
// 啓動程序
func StartWebSocket() {
    http.HandleFunc("/acc", wsPage)
    http.ListenAndServe(":8089", nil)
}

3.1.2 升級協議

  • 客戶端是經過http請求發送到服務端，咱們須要對http協議進行升級爲websocket協議
  • 對http請求協議進行升級 golang 庫gorilla/websocket 已經作得很好了，咱們直接使用就能夠了
  • 在實際使用的時候，建議每一個鏈接使用兩個協程處理客戶端請求數據和向客戶端發送數據，雖然開啓協程會佔用一些內存，可是讀取分離，減小收發數據堵塞的可能
  • init_acc.go
func wsPage(w http.ResponseWriter, req *http.Request) {

    // 升級協議
    conn, err := (&websocket.Upgrader{CheckOrigin: func(r *http.Request) bool {
        fmt.Println("升級協議", "ua:", r.Header["User-Agent"], "referer:", r.Header["Referer"])

        return true
    }}).Upgrade(w, req, nil)
    if err != nil {
        http.NotFound(w, req)

        return
    }

    fmt.Println("webSocket 創建鏈接:", conn.RemoteAddr().String())

    currentTime := uint64(time.Now().Unix())
    client := NewClient(conn.RemoteAddr().String(), conn, currentTime)

    go client.read()
    go client.write()

    // 用戶鏈接事件
    clientManager.Register <- client
}

3.1.3 客戶端鏈接的管理

  • 當前程序有多少用戶鏈接，還須要對用戶廣播的須要，這裏咱們就須要一個管理者(clientManager)，處理這些事件:
  • 記錄所有的鏈接、登陸用戶的能夠經過 appId+uuid 查到用戶鏈接
  • 使用map存儲，就涉及到多協程併發讀寫的問題，因此須要加讀寫鎖
  • 定義四個channel ，分別處理客戶端創建鏈接、用戶登陸、斷開鏈接、全員廣播事件
// 鏈接管理
type ClientManager struct {
    Clients     map[*Client]bool   // 所有的鏈接
    ClientsLock sync.RWMutex       // 讀寫鎖
    Users       map[string]*Client // 登陸的用戶 // appId+uuid
    UserLock    sync.RWMutex       // 讀寫鎖
    Register    chan *Client       // 鏈接鏈接處理
    Login       chan *login        // 用戶登陸處理
    Unregister  chan *Client       // 斷開鏈接處理程序
    Broadcast   chan []byte        // 廣播 向所有成員發送數據
}

// 初始化
func NewClientManager() (clientManager *ClientManager) {
    clientManager = &ClientManager{
        Clients:    make(map[*Client]bool),
        Users:      make(map[string]*Client),
        Register:   make(chan *Client, 1000),
        Login:      make(chan *login, 1000),
        Unregister: make(chan *Client, 1000),
        Broadcast:  make(chan []byte, 1000),
    }

    return
}

3.1.4 註冊客戶端的socket的寫的異步處理程序

  • 防止發生程序崩潰，因此須要捕獲異常
  • 爲了顯示異常崩潰位置這裏使用string(debug.Stack())打印調用堆棧信息
  • 若是寫入數據失敗了，可能鏈接有問題，就關閉鏈接
  • client.go
// 向客戶端寫數據
func (c *Client) write() {
    defer func() {
        if r := recover(); r != nil {
            fmt.Println("write stop", string(debug.Stack()), r)

        }
    }()

    defer func() {
        clientManager.Unregister <- c
        c.Socket.Close()
        fmt.Println("Client發送數據 defer", c)
    }()

    for {
        select {
        case message, ok := <-c.Send:
            if !ok {
                // 發送數據錯誤 關閉鏈接
                fmt.Println("Client發送數據 關閉鏈接", c.Addr, "ok", ok)

                return
            }

            c.Socket.WriteMessage(websocket.TextMessage, message)
        }
    }
}

3.1.5 註冊客戶端的socket的讀的異步處理程序

  • 循環讀取客戶端發送的數據並處理
  • 若是讀取數據失敗了，關閉channel
  • client.go
// 讀取客戶端數據
func (c *Client) read() {
    defer func() {
        if r := recover(); r != nil {
            fmt.Println("write stop", string(debug.Stack()), r)
        }
    }()

    defer func() {
        fmt.Println("讀取客戶端數據 關閉send", c)
        close(c.Send)
    }()

    for {
        _, message, err := c.Socket.ReadMessage()
        if err != nil {
            fmt.Println("讀取客戶端數據 錯誤", c.Addr, err)

            return
        }

        // 處理程序
        fmt.Println("讀取客戶端數據 處理:", string(message))
        ProcessData(c, message)
    }
}

3.1.6 接收客戶端數據並處理

  • 約定發送和接收請求數據格式，爲了js處理方便，採用了json的數據格式發送和接收數據(人類能夠閱讀的格式在工做開發中使用是比較方便的)
  • 登陸發送數據示例:
{"seq":"1565336219141-266129","cmd":"login","data":{"userId":"馬遠","appId":101}}
  • 登陸響應數據示例:
{"seq":"1565336219141-266129","cmd":"login","response":{"code":200,"codeMsg":"Success","data":null}}
  • websocket是雙向的數據通信，能夠連續發送，若是發送的數據須要服務端回覆，就須要一個seq來肯定服務端的響應是回覆哪一次的請求數據
  • cmd 是用來肯定動做，websocket沒有相似於http的url,因此規定 cmd 是什麼動做
  • 目前的動做有:login/heartbeat 用來發送登陸請求和鏈接保活(長時間沒有數據發送的長鏈接容易被瀏覽器、移動中間商、nginx、服務端程序斷開)
  • 爲何須要AppId,UserId是表示用戶的惟一字段，設計的時候爲了作成通用性，設計AppId用來表示用戶在哪一個平臺登陸的(web、app、ios等)，方便後續擴展
  • request_model.go 約定的請求數據格式
/************************  請求數據  **************************/
// 通用請求數據格式
type Request struct {
    Seq  string      `json:"seq"`            // 消息的惟一Id
    Cmd  string      `json:"cmd"`            // 請求命令字
    Data interface{} `json:"data,omitempty"` // 數據 json
}

// 登陸請求數據
type Login struct {
    ServiceToken string `json:"serviceToken"` // 驗證用戶是否登陸
    AppId        uint32 `json:"appId,omitempty"`
    UserId       string `json:"userId,omitempty"`
}

// 心跳請求數據
type HeartBeat struct {
    UserId string `json:"userId,omitempty"`
}
  • response_model.go
/************************  響應數據  **************************/
type Head struct {
    Seq      string    `json:"seq"`      // 消息的Id
    Cmd      string    `json:"cmd"`      // 消息的cmd 動做
    Response *Response `json:"response"` // 消息體
}

type Response struct {
    Code    uint32      `json:"code"`
    CodeMsg string      `json:"codeMsg"`
    Data    interface{} `json:"data"` // 數據 json
}

3.1.7 使用路由的方式處理客戶端的請求數據

  • 使用路由的方式處理由客戶端發送過來的請求數據
  • 之後添加請求類型之後就能夠用類是用http相相似的方式(router-controller)去處理
  • acc_routers.go
// Websocket 路由
func WebsocketInit() {
    websocket.Register("login", websocket.LoginController)
    websocket.Register("heartbeat", websocket.HeartbeatController)
}

3.1.8 防止內存溢出和Goroutine不回收

    1. 定時任務清除超時鏈接

沒有登陸的鏈接和登陸的鏈接6分鐘沒有心跳則斷開鏈接

client_manager.go

// 定時清理超時鏈接
func ClearTimeoutConnections() {
    currentTime := uint64(time.Now().Unix())

    for client := range clientManager.Clients {
        if client.IsHeartbeatTimeout(currentTime) {
            fmt.Println("心跳時間超時 關閉鏈接", client.Addr, client.UserId, client.LoginTime, client.HeartbeatTime)

            client.Socket.Close()
        }
    }
}
    1. 讀寫的Goroutine有一個失敗，則相互關閉

write()Goroutine寫入數據失敗，關閉c.Socket.Close()鏈接，會關閉read()Goroutine
read()Goroutine讀取數據失敗，關閉close(c.Send)鏈接，會關閉write()Goroutine

    1. 客戶端主動關閉

關閉讀寫的Goroutine
從ClientManager刪除鏈接

    1. 監控用戶鏈接、Goroutine數

十個內存溢出有九個和Goroutine有關
添加一個http的接口，能夠查看系統的狀態，防止Goroutine不回收
查看系統狀態

    1. Nginx 配置不活躍的鏈接釋放時間，防止忘記關閉的鏈接
    1. 使用 pprof 分析性能、耗時

3.2 使用javaScript實現webSocket客戶端

3.2.1 啓動並註冊監聽程序

  • js 創建鏈接，並處理鏈接成功、收到數據、斷開鏈接的事件處理
ws = new WebSocket("ws://127.0.0.1:8089/acc");

 
ws.onopen = function(evt) {
  console.log("Connection open ...");
};
 
ws.onmessage = function(evt) {
  console.log( "Received Message: " + evt.data);
  data_array = JSON.parse(evt.data);
  console.log( data_array);
};
 
ws.onclose = function(evt) {
  console.log("Connection closed.");
};

3.2.2 發送數據

  • 須要注意:鏈接創建成功之後才能夠發送數據
  • 創建鏈接之後由客戶端向服務器發送數據示例
登陸:
ws.send('{"seq":"2323","cmd":"login","data":{"userId":"11","appId":101}}');

心跳:
ws.send('{"seq":"2324","cmd":"heartbeat","data":{}}');
 
關閉鏈接:
ws.close();

四、goWebSocket 項目

4.1 項目說明

  • 本項目是基於webSocket實現的分佈式IM系統
  • 客戶端隨機分配用戶名，全部人進入一個聊天室，實現羣聊的功能
  • 單臺機器(24核128G內存)支持百萬客戶端鏈接
  • 支持水平部署，部署的機器之間能夠相互通信
  • 項目架構圖

網站架構圖

4.2 項目依賴

  • 本項目只須要使用 redis 和 golang
  • 本項目使用govendor管理依賴，克隆本項目就能夠直接使用
# 主要使用到的包
github.com/gin-gonic/gin@v1.4.0
github.com/go-redis/redis
github.com/gorilla/websocket
github.com/spf13/viper
google.golang.org/grpc
github.com/golang/protobuf

4.3 項目啓動

  • 克隆項目
git clone git@github.com:link1st/gowebsocket.git
# 或
git clone https://github.com/link1st/gowebsocket.git
  • 修改項目配置
cd gowebsocket
cd config
mv app.yaml.example app.yaml
# 修改項目監聽端口，redis鏈接等(默認127.0.0.1:3306)
vim app.yaml
# 返回項目目錄，爲之後啓動作準備
cd ..
  • 配置文件說明
app:
  logFile: log/gin.log # 日誌文件位置
  httpPort: 8080 # http端口
  webSocketPort: 8089 # webSocket端口
  rpcPort: 9001 # 分佈式部署程序內部通信端口
  httpUrl: 127.0.0.1:8080
  webSocketUrl:  127.0.0.1:8089


redis:
  addr: "localhost:6379"
  password: ""
  DB: 0
  poolSize: 30
  minIdleConns: 30
  • 啓動項目
go run main.go
  • 進入IM聊天地址

http://127.0.0.1:8080/home/index

  • 到這裏，就能夠體驗到基於webSocket的IM系統

五、webSocket項目Nginx配置

5.1 爲何要配置Nginx

  • 使用nginx實現內外網分離，對外只暴露Nginx的Ip(通常的互聯網企業會在nginx以前加一層LVS作負載均衡)，減小入侵的可能
  • 使用Nginx能夠利用Nginx的負載功能，前端再使用的時候只須要鏈接固定的域名，經過Nginx將流量分發了到不一樣的機器
  • 同時咱們也可使用Nginx的不一樣的負載策略(輪詢、weight、ip_hash)

5.2 nginx配置

  • 使用域名 im.91vh.com 爲示例，參考配置
  • 一級目錄im.91vh.com/acc 是給webSocket使用，是用nginx stream轉發功能(nginx 1.3.31 開始支持，使用Tengine配置也是相同的)，轉發到golang 8089 端口處理
  • 其它目錄是給HTTP使用，轉發到golang 8080 端口處理
upstream  go-im
{
    server 127.0.0.1:8080 weight=1 max_fails=2 fail_timeout=10s;
    keepalive 16;
}

upstream  go-acc
{
    server 127.0.0.1:8089 weight=1 max_fails=2 fail_timeout=10s;
    keepalive 16;
}


server {
    listen       80 ;
    server_name  im.91vh.com;
    index index.html index.htm ;


    location /acc {
        proxy_set_header Host $host;
        proxy_pass http://go-acc;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        proxy_set_header Connection "";
        proxy_redirect off;
        proxy_intercept_errors on;
        client_max_body_size 10m;
    }

    location /
    {
        proxy_set_header Host $host;
        proxy_pass http://go-im;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        proxy_redirect off;
        proxy_intercept_errors on;
        client_max_body_size 30m;
    }

    access_log  /link/log/nginx/access/im.log;
    error_log   /link/log/nginx/access/im.error.log;
}

5.3 問題處理

  • 運行nginx測試命令，查看配置文件是否正確
/link/server/tengine/sbin/nginx -t
  • 若是出現錯誤
nginx: [emerg] unknown "connection_upgrade" variable
configuration file /link/server/tengine/conf/nginx.conf test failed
  • 處理方法
  • 在nginx.com添加
http{
    fastcgi_temp_file_write_size 128k;
..... # 須要添加的內容

    #support websocket
    map $http_upgrade $connection_upgrade {
        default upgrade;
        ''      close;
    }

.....
    gzip on;
    
}
  • 緣由:Nginx代理webSocket的時候就會遇到Nginx的設計問題 End-to-end and Hop-by-hop Headers

六、壓測

6.1 Linux內核優化

  • 設置文件打開句柄數
ulimit -n 1000000
  • 設置sockets鏈接參數
vim /etc/sysctl.conf
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_tw_recycle = 0

6.2 壓測準備

  • 待壓測，若是你們有壓測的結果歡迎補充

6.3 壓測數據

  • 項目在實際使用的時候，每一個鏈接約佔 24Kb內存，一個Goroutine 約佔11kb
  • 支持百萬鏈接須要22G內存
在線用戶數 cup 內存 I/O net.out
1W
10W
100W

七、如何基於webSocket實現一個分佈式Im

7.1 說明

獲取所有在線的用戶，查詢單前服務的所有用戶+集羣中服務的所有用戶
發送消息，這裏採用的是http接口發送(微信網頁版發送消息也是http接口)，這裏考慮主要是兩點:
1.服務分離，讓acc系統儘可能的簡單一點，不摻雜其它業務邏輯
2.發送消息是走http接口，不使用webSocket鏈接，才用收和發送數據分離的方式，能夠加快收發數據的效率

7.2 架構

  • 項目啓動註冊和用戶鏈接時序圖

用戶鏈接時序圖

  • 其它系統(IM、任務)向webSocket(acc)系統鏈接的用戶發送消息時序圖

分佈是系統隨機給用戶發送消息

八、回顧和反思

8.1 在其它系統應用

  • 本系統設計的初衷就是:和客戶端保持一個長連接、對外部系統兩個接口(查詢用戶是否在線、給在線的用戶推送消息)，實現業務的分離
  • 只有和業務分離可，才能夠供多個業務使用，而不是每一個業務都創建一個長連接

8.2 已經實現的功能

  • gin log日誌(請求日誌+debug日誌)
  • 讀取配置文件 完成
  • 定時腳本，清理過時未心跳連接 完成
  • http接口，獲取登陸、連接數量 完成
  • http接口，發送push、查詢有多少人在線 完成
  • grpc 程序內部通信，發送消息 完成
  • appIds 一個用戶在多個平臺登陸
  • 界面，把全部在線的人拉倒一個羣裏面，發送消息 完成
  • 單聊、羣聊 完成
  • 實現分佈式，水平擴張 完成
  • 壓測腳本
  • 文檔整理
  • 文檔目錄、百萬長連接的實現、爲何要實現一個IM、怎麼實現一個Im
  • 架構圖以及擴展

IM實現細節:

  • 定義文本消息結構 完成
  • html發送文本消息 完成
  • 接口接收文本消息併發送給全體 完成
  • html接收到消息 顯示到界面 完成
  • 界面優化 須要持續優化
  • 有人加入之後廣播全體 完成
  • 定義加入聊天室的消息結構 完成
  • 引入機器人 待定

8.2 須要完善、優化

  • 登陸，使用微信登陸 獲取暱稱、頭像等
  • 有帳號系統、資料系統
  • 界面優化、適配手機端
  • 消息 文本消息(支持表情)、圖片、語音、視頻消息
  • 微服務註冊、發現、熔斷等
  • 添加配置項，單臺機器最大鏈接數量

8.3 總結

  • 雖然實現了一個分佈式在聊天的IM，可是有不少細節沒有處理(登陸沒有鑑權、界面還待優化等)，可是能夠經過這個示例能夠了解到:經過WebSocket解決不少業務上需求
  • 本文雖然號稱單臺機器能有百萬長連接(內存上能知足)，可是實際在場景遠比這個複雜(cpu有些壓力)，固然了若是你有這麼大的業務量能夠購買更多的機器更好的去支撐你的業務，本程序只是演示如何在實際工做用使用webSocket.
  • 參考本文，你能夠實現出來符合你須要的程序

九、參考文獻

維基百科 WebSocket

阮一峯 WebSocket教程

WebSocket協議：5分鐘從入門到精通

link1st gowebsocket

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程