基于websocket单台机器支持百万连接分布式聊天(IM)系统

本文将介绍如何实现一个基于websocket分布式聊天(IM)系统。

使用golang实现websocket通讯，单机可以支持百万连接，使用gin框架、nginx负载、可以水平部署、程序内部相互通讯、使用grpc通讯协议。

本文内容比较长，如果直接想clone项目体验直接进入项目体验 goWebSocket项目下载 ,文本从介绍webSocket是什么开始，然后开始介绍这个项目，以及在Nginx中配置域名做webSocket的转发，然后介绍如何搭建一个分布式系统。

目录

• 1、项目说明
  • 1.1 goWebSocket
  • 1.2 项目体验
• 2、介绍webSocket
  • 2.1 webSocket 是什么
  • 2.2 webSocket的兼容性
  • 2.3 为什么要用webSocket
  • 2.4 webSocket建立过程
• 3、如何实现基于webSocket的长连接系统
  • 3.1 使用go实现webSocket服务端
    • 3.1.1 启动端口监听
    • 3.1.2 升级协议
    • 3.1.3 客户端连接的管理
    • 3.1.4 注册客户端的socket的写的异步处理程序
    • 3.1.5 注册客户端的socket的读的异步处理程序
    • 3.1.6 接收客户端数据并处理
    • 3.1.7 使用路由的方式处理客户端的请求数据
    • 3.1.8 防止内存溢出和Goroutine不回收
  • 3.2 使用javaScript实现webSocket客户端
    • 3.2.1 启动并注册监听程序
    • 3.2.2 发送数据
  • 3.3 发送消息
    • 3.3.1 文本消息
    • 3.3.2 图片和语言消息
• 4、goWebSocket 项目
  • 4.1 项目说明
  • 4.2 项目依赖
  • 4.3 项目启动
  • 4.4 接口文档
    • 4.4.1 HTTP接口文档
      • 4.4.1.1 接口说明
      • 4.4.1.2 聊天页面
      • 4.4.1.3 获取房间用户列表
      • 4.4.1.4 查询用户是否在线
      • 4.4.1.5 给用户发送消息
      • 4.4.1.6 给全员用户发送消息
    • 4.4.2 RPC接口文档
      • 4.4.2.1 接口说明
      • 4.4.2.2 查询用户是否在线
      • 4.4.2.3 发送消息
      • 4.4.2.4 给指定房间所有用户发送消息
      • 4.4.2.5 获取房间内全部用户
• 5、webSocket项目Nginx配置
  • 5.1 为什么要配置Nginx
  • 5.2 nginx配置
  • 5.3 问题处理
• 6、压测
  • 6.1 Linux内核优化
  • 6.2 压测准备
  • 6.3 压测数据
• 7、如何基于webSocket实现一个分布式Im
  • 7.1 说明
  • 7.2 架构
  • 7.3 分布式系统部署
• 8、回顾和反思
  • 8.1 在其它系统应用
  • 8.2 需要完善、优化
  • 8.3 总结
• 9、参考文献

1、项目说明

1.1 goWebSocket

本文将介绍如何实现一个基于websocket聊天(IM)分布式系统。

使用golang实现websocket通讯，单机支持百万连接，使用gin框架、nginx负载、可以水平部署、程序内部相互通讯、使用grpc通讯协议。

• 一般项目中webSocket使用的架构图

1.2 项目体验

• 项目地址 gowebsocket
• IM-聊天首页 或者在新的窗口打开 http://im.91vh.com/home/index
• 打开连接以后进入聊天界面
• 多人群聊可以同时打开两个窗口

2、介绍webSocket

2.1 webSocket 是什么

WebSocket 协议在2008年诞生，2011年成为国际标准。所有浏览器都已经支持了。

它的最大特点就是，服务器可以主动向客户端推送信息，客户端也可以主动向服务器发送信息，是真正的双向平等对话，属于服务器推送技术的一种。

• HTTP和WebSocket在通讯过程的比较

• HTTP和webSocket都支持配置证书，ws:// 无证书 wss:// 配置证书的协议标识

2.2 webSocket的兼容性

• 浏览器的兼容性，开始支持webSocket的版本

• 服务端的支持

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

• Android和IOS的支持

Android可以使用java-webSocket对webSocket支持

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

2.3 为什么要用webSocket

• 1. 从业务上出发，需要一个主动通达客户端的能力

     目前大多数的请求都是使用HTTP，都是由客户端发起一个请求，有服务端处理，然后返回结果，不可以服务端主动向某一个客户端主动发送数据

- 2. 大多数场景我们需要主动通知用户，如:聊天系统、用户完成任务主动告诉用户、一些运营活动需要通知到在线的用户 - 3. 可以获取用户在线状态 - 4. 在没有长连接的时候通过客户端主动轮询获取数据 - 5. 可以通过一种方式实现，多种不同平台(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

• 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. 升级协议完成以后，客户端和服务器就可以相互发送数据

3、如何实现基于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":{}}');

ping 查看服务是否正常:
ws.send('{"seq":"2325","cmd":"ping","data":{}}');

关闭连接:
ws.close();

3.3 发送消息

3.3.1 文本消息

客户端只要知道发送用户是谁，还有内容就可以显示文本消息，这里我们重点关注一下数据部分

target：定义接收的目标，目前未设置

type：消息的类型，text 文本消息 img 图片消息

msg：文本消息内容

from：消息的发送者

文本消息的结构:

{
  "seq": "1569080188418-747717",
  "cmd": "msg",
  "response": {
    "code": 200,
    "codeMsg": "Ok",
    "data": {
      "target": "",
      "type": "text",
      "msg": "hello",
      "from": "马超"
    }
  }
}

这样一个文本消息的结构就设计完成了，客户端在接收到消息内容就可以展现到 IM 界面上

3.3.2 图片和语言消息

发送图片消息，发送消息者的客户端需要先把图片上传到文件服务器，上传成功以后获得图片访问的 URL，然后由发送消息者的客户端需要将图片 URL 发送到 gowebsocket，gowebsocket 图片的消息格式发送给目标客户端，消息接收者客户端接收到图片的 URL 就可以显示图片消息。

图片消息的结构:

{
  "type": "img",
  "from": "马超",
  "url": "http://91vh.com/images/home_logo.png",
  "secret": "消息鉴权 secret",
  "size": {
    "width": 480,
    "height": 720
  }
}

语言消息、和视频消息和图片消息类似，都是先把文件上传服务器，然后通过 gowebsocket 传递文件的 URL，需要注意的是部分消息涉及到隐私的文件，文件访问的时候需要做好鉴权信息，不能让非接收用户也能查看到别人的消息内容。

4、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系统

4.4 接口文档

4.4.1.1 接口说明