一、WebSocket基本观点先容
1、WebSocket是什么？
WebSocket 协议在2008年出身，2011年景为国际标准。所有浏览器都已经支持了。
它的最大特点便是，做事器能够主动向客户端推送信息，客户端也能够主动向做事器发送信息，是真正的双向平等对话，属于目前做事器推送技能的个中一种。
HTTP和WebSocket在通讯过程的比较HTTP和webSocket都支持配置证书，ws:// 无证书 wss:// 配置证书的协议标识一样平常项目中webSocket利用的架构图2、WebSocket的兼容性
目前已经支持webSocket的浏览器版本做事端技能的支持情形golang、java、php、node.js、python、nginx 都有不错的支持。
Android系统和IOS系统的支持Android系统可以利用java-webSocket对webSocket来支持。
iOS 4.2及以上版本均具有WebSockets支持。
3、为什么要用WebSocket？
(1)从业务角度考虑，须要一个主动通达客户真个能力
目前大多数的要求都是利用HTTP，都是由客户端发起一个要求，有做事端处理，然后返回结果，不可以做事端主动向某一个客户端主动发送数据；
(2)大多数场景我们须要主动关照用户，如:谈天系统、用户完成任务主动见告用户、一些运营活动须要关照到在线的用户；
(3)可以获取用户在线状态；
(4)在没有长连接的时候通过客户端主动轮询获取数据；
(5)可以通过一种办法实现，多种不同平台(H5/Android/IOS)去利用。
4、WebSocke的建立过程
(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
(2)做事器相应升级协议
做事端吸收到升级协议的要求，如果做事端支持升级协议会做如下相应
返回:
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
(3)升级协议完成往后，客户端和做事器就可以相互发送数据
二、如何基于webSocket实现长连接系统
1、利用go实现webSocket做事端
(1) 启动端口监听
websocket须要监听端口，因此须要在golang 成功的 main 函数中用协程的办法去启动程序main.go 实现启动go websocket.StartWebSocket()
init_acc.go 启动程序// 启动程序
func StartWebSocket() {
http.HandleFunc("/acc", wsPage)
http.ListenAndServe(":8089", nil)
}
(2) 升级的协议
客户端是通过http要求发送到做事端，我们须要对http协议进行升级为websocket协议；对http要求协议进行升级 golang 库gorilla/websocket 比较成熟，直策应用即可；在实际利用的时候，建议每个连接利用两个协程处理客户端要求数据和向客户端发送数据，虽然开启协程会占用一些内存，但是读取分离，减少收发数据堵塞的可能；init_acc.gofunc 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) 客户端连接的管理
当出路序有多少用户连接，还须要对用户广播的须要，这里我们就须要一个管理者(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
}
(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)
}
}
}
(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)
}
}
(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
}
(7)利用路由的办法处理客户真个要求数据
利用路由的办法处理由客户端发送过来的要求数据往后添加要求类型往后就可以用类是用http相类似的办法(router-controller)去处理acc_routers.go// Websocket 路由
func WebsocketInit() {
websocket.Register("login", websocket.LoginController)
websocket.Register("heartbeat", websocket.HeartbeatController)
}
(8) 防止内存溢出和Goroutine不回收
a、定时任务打消超时连接 没有登录的连接和登录的连接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()
}
}
}
b、读写的Goroutine有一个失落败，则相互关闭 write()Goroutine写入数据失落败，关闭c.Socket.Close()连接，会关闭read()Goroutine read()Goroutine读取数据失落败，关闭close(c.Send)连接，会关闭write()Goroutine
c、客户端主动关闭 关闭读写的Goroutine 从ClientManager删除连接
d、监控用户连接、Goroutine数 十个内存溢出有九个和Goroutine有关 添加一个http的接口，可以查看系统的状态，防止Goroutine不回收 查看系统状态
e、Nginx 配置不生动的连接开释韶光，防止忘却关闭的连接
f、利用 pprof 剖析性能、耗时
2、利用javaScript实现webSocket客户端
(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.");
};
(2) 发送数据
须要把稳:连接建立成功往后才可以发送数据建立连接往后由客户端向做事器发送数据示例登录:
ws.send('{"seq":"2323","cmd":"login","data":{"userId":"11","appId":101}}');
心跳:
ws.send('{"seq":"2324","cmd":"heartbeat","data":{}}');
关闭连接:
ws.close();
三、goWebSocket 项目
1、项目解释
本项目是基于webSocket实现的分布式谈天系统客户端随机分配用户名，所有人进入一个谈天室，实现群聊的功能单台机器(24核128G内存)支持百万客户端连接支持水平支配，支配的机器之间可以相互通讯项目架构图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
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配置
1、为什么要配置Nginx？
利用nginx实现内外网分离，对外只暴露Nginx的Ip(一样平常的互联网企业会在nginx之前加一层LVS做负载均衡)，减少入侵的可能；利用Nginx可以利用Nginx的负载功能，前端再利用的时候只须要连接固定的域名，通过Nginx将流量分发了到不同的机器；同时我们也可以利用Nginx的不同的负载策略(轮询、weight、ip_hash)。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;
}
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五、压测
1、Linux内核参数的优化
设置文件打开句柄数ulimit -n 1000000
设置sockets连接参数vim /etc/sysctl.conf
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_tw_recycle = 0
2、压测准备
待压测，如果大家有压测的结果欢迎补充3、 压测数据
项目在实际利用的时候，每个连接约占 24Kb内存，一个Goroutine 约占11kb支持百万连接须要22G内存六、如何基于webSocket实现一个分布式谈天系统？
1、解释
参考本项目源码gowebsocket v1.0.0 单机版谈天系统；gowebsocket v1.0.0 分布式谈天系统；为了方便演示，IM系统和webSocket(acc)系统合并在一个别系中；IM系统接口: 获取全部在线的用户，查询单前做事的全部用户+集群中做事的全部用户 发送，这里采取的是http接口发送(微信网页版发送也是http接口)，这里考虑紧张是两点: 1.做事分离，让acc系统只管即便的大略一点，不掺杂其它业务逻辑 2.发送是走http接口，不该用webSocket连接，才用收和发送数据分离的办法，可以加快收发数据的效率；2、架构
项目启动注册和用户连接时序图其它系统(IM、任务)向webSocket(acc)系统连接的用户发送时序图
七、总结与回顾
1、在其它业务系统上的运用
本系统设计的初衷便是:和客户端保持一个长连接、对外部系统两个接口(查询用户是否在线、给在线的用户推送)，实现业务的分离；只有和业务分离开，才可以供多个业务利用，而不是每个业务都建立一个长链接。2、 已经实现的功能
gin log日志(要求日志+debug日志)读取配置文件 完成定时脚本，清理过期未心跳连接 完成http接口，获取登录、连接数量 完成http接口，发送push、查询有多少人在线 完成grpc 程序内部通讯，发送 完成appIds 一个用户在多个平台登录界面，把所有在线的人拉倒一个群里面，发送 完成单聊、群聊 完成实现分布式，水平扩展 完成压测脚本文档整理架构图以及扩展。IM实现细节:
定义文本构造 完成；html发送文本 完成；接口吸收文本并发送给全体 完成；html吸收到 显示到界面 完成；界面优化 须要持续优化；有人加入往后广播全体 完成；定义加入谈天室的构造 完成；引入机器人 待定。3、还须要连续完善和优化的地方
登录，利用微信登录 获取昵称、头像等；有账号系统、资料系统；界面优化、适配手机端； 文本(支持表情)、图片、语音、视频；微做事注册、创造、熔断等；添加配置项，单台机器最大连接数量。4、 总结
虽然实现了一个分布式谈天系统，但是有很多细节没有处理(登录没有鉴权、界面还待优化等)，但是可以通过这个示例可以理解到:通过WebSocket办理很多业务上需求；本文虽然号称单台机器能有百万长连接(内存上能知足)，但是实际在场景远比这个繁芜(cpu有些压力)，当然了如果你有这么大的业务量可以购买更多的机器更好的去支撑你的业务，本程序只是演示如何在实际工浸染利用webSocket；参考本文章，你就可以实现出来符合你须要的谈天程序。如果你喜好本文章，请帮忙关注、转发、点赞，本人将定期更新一些技能“干货”分享给大家，希望能帮助大家办理事情中的一些问题，感激！
