[网络协议]Golang/Gin-WebSocket实现实时消息推送

前言

WebSocket在 HTML5 游戏和网页消息推送都使用比较多。

WebSocket 是 HTML5 的重要特性，它实现了基于浏览器的远程socket，它使浏览器和服务器可以进行全双工通信。

目前Go中用的比较多的WebSocket包是gorilla/websocket，本文将介绍如何使用gorilla/websocket，在Gin框架下编写WebSocket实时消息推送。

gorilla/websocket基础用法

gorilla/websocket相关链接：

Github地址：https://github.com/gorilla/websocket

文档：https://godoc.org/github.com/gorilla/websocket

gorilla/websocket的安装：

go get github.com/gorilla/websocket

Upgrader用于升级 http 请求，把 http 请求升级为长连接的 WebSocket。

Hello World示例：

package main

import (
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	"log"
	"net/http"
	"time"
)

var upgrader = websocket.Upgrader{
  // 这个是校验请求来源
  // 在这里我们不做校验，直接return true
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

func main() {
	engine := gin.Default()

	engine.GET("/helloWebSocket", func(context *gin.Context) {
    // 将普通的http GET请求升级为websocket请求
		client, _ := upgrader.Upgrade(context.Writer, context.Request, nil)
		for {
			// 每隔两秒给前端推送一句消息“hello, WebSocket”
			err := client.WriteMessage(websocket.TextMessage, []byte("hello, WebSocket"))
			if err != nil {
				log.Println(err)
			}
			time.Sleep(time.Second * 2)
		}
	})

	err := engine.Run(":8090")
	if err != nil {
		log.Fatalln(err)
	}
}

写完以后你可以用websocket在线测试工具测试你的代码：http://coolaf.com/tool/chattest。

请求url前面的http://记得换成ws://。

实现实时消息推送

这部分我参考了GitHub上lwnmengjing的代码，自己做了一些修改以贴合自己的需求，大家也可以直接到https://github.com/lwnmengjing/pushMessage参考他的代码。

代码

以下是我自己修改后的代码：

package module

import (
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	"log"
	"net/http"
	"sync"
	"time"
)

var (
  // 消息通道
	news = make(map[string]chan interface{})
  // websocket客户端链接池
	client = make(map[string]*websocket.Conn)
  // 互斥锁，防止程序对统一资源同时进行读写
	mux sync.Mutex
)

// api:/getPushNews接口处理函数
func GetPushNews(context *gin.Context)  {
	id := context.Query("userId")
	log.Println(id + "websocket链接")
  // 升级为websocket长链接
	WsHandler(context.Writer, context.Request, id)
}

// api:/deleteClient接口处理函数
func DeleteClient(context *gin.Context)  {
	id := context.Param("id")
  // 关闭websocket链接
	conn, exist := getClient(id)
	if exist {
		conn.Close()
		deleteClient(id)
	} else {
		context.JSON(http.StatusOK, gin.H{
			"mesg": "未找到该客户端",
		})
	}
  // 关闭其消息通道
	_, exist =getNewsChannel(id)
	if exist {
		deleteNewsChannel(id)
	}
}

// websocket Upgrader
var wsupgrader = websocket.Upgrader{
	ReadBufferSize:   1024,
	WriteBufferSize:  1024,
	HandshakeTimeout: 5 * time.Second,
	// 取消ws跨域校验
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

// WsHandler 处理ws请求
func WsHandler(w http.ResponseWriter, r *http.Request, id string)  {
	var conn *websocket.Conn
	var err error
	var exist bool
  // 创建一个定时器用于服务端心跳
	pingTicker := time.NewTicker(time.Second * 10)
	conn, err = wsupgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println(err)
		return
	}
	// 把与客户端的链接添加到客户端链接池中
	addClient(id, conn)

	// 获取该客户端的消息通道
	m, exist := getNewsChannel(id)
	if !exist {
		m = make(chan interface{})
		addNewsChannel(id, m)
	}

	// 设置客户端关闭ws链接回调函数
	conn.SetCloseHandler(func(code int, text string) error {
		deleteClient(id)
		log.Println(code)
		return nil
	})

	for {
		select {
		case content, _ := <- m:
      // 从消息通道接收消息，然后推送给前端
			err = conn.WriteJSON(content)
			if err != nil {
				log.Println(err)
				conn.Close()
				deleteClient(id)
				return
			}
		case <- pingTicker.C:
      // 服务端心跳:每20秒ping一次客户端，查看其是否在线
			conn.SetWriteDeadline(time.Now().Add(time.Second * 20))
			err = conn.WriteMessage(websocket.PingMessage, []byte{})
			if err != nil {
				log.Println("send ping err:", err)
				conn.Close()
				deleteClient(id)
				return
			}
		}
	}
}

// 将客户端添加到客户端链接池
func addClient(id string, conn *websocket.Conn) {
	mux.Lock()
	client[id] = conn
	mux.Unlock()
}

// 获取指定客户端链接
func getClient(id string) (conn *websocket.Conn, exist bool) {
	mux.Lock()
	conn, exist = client[id]
	mux.Unlock()
	return
}

// 删除客户端链接
func deleteClient(id string) {
	mux.Lock()
	delete(client, id)
	log.Println(id + "websocket退出")
	mux.Unlock()
}

// 添加用户消息通道
func addNewsChannel(id string, m chan interface{}) {
	mux.Lock()
	news[id] = m
	mux.Unlock()
}

// 获取指定用户消息通道
func getNewsChannel(id string) (m chan interface{}, exist bool) {
	mux.Lock()
	m, exist = news[id]
	mux.Unlock()
	return
}

// 删除指定消息通道
func deleteNewsChannel(id string) {
	mux.Lock()
	if m, ok := news[id]; ok {
		close(m)
		delete(news, id)
	}
	mux.Unlock()
}

api

补充说明

当你要给某个用户推送消息时，你只需要使用getNewsChannel()方法获取该用户的消息通道，然后把消息送入通道就可以了。

若用户离线，你可以把消息直接存到用户所有消息中，或者设置一个消息队列，把消息放到用户未读消息队列中，下次用户上线时再一次性推送给用户。

服务端心跳：服务端每隔20秒回ping一下用户，查看其是否还在线，若ping不到，则服务端自动关闭websocket链接。