鉴于聊天已然成为大部分app的基础功能,而大部分app用户基数有没有辣么大,常用的聊天server架构如xmpp或者消息队列实现之类的用起来还挺麻烦的,有比较难跟网页端做交互,加之H5标准落地,所以websocket已然成为一个轻巧可用性高的聊天server实现方法;
websocket的server常见的是用nodejs或者java的netty框架实现,netty相对重一点,direct buffer的内存泄露调起来比较麻烦,试了一下go,轻巧,稳定性不错,性能不错,所以用go实现了一下;
websocket的协议标准和基本概念网上一搜一片,这里不赘述;
http server用gin来做,websocket的handler则用gorilla,由于不重复造轮子,所以整个搭建的过程很快;
import ( "util" "os" "fmt" "github.com/DeanThompson/ginpprof" "github.com/gin-gonic/gin" "runtime" ) var ( logger * util.LogHelper ) func main() { runtime.GOMAXPROCS(runtime.NumCPU()) logFile,err := os.OpenFile("/var/log/gows.log",os.O_CREATE|os.O_RDWR,0777) if err!=nil { fmt.Println(err.Error()) os.Exit(0) } defer logFile.Close() logger = util.NewLogger(logFile) logger.Info("Starting system...") wsHandler := new(WebSocketHandler) gin.SetMode(gin.ReleaseMode) r := gin.Default() r.GET("/", func(c *gin.Context) { wsHandler.HandleConn(c.Writer, c.Request) }) ginpprof.Wrapper(r)//调试用 可以看到堆栈状态和所有goroutine状态 //err = r.Run(listenPath, certPath, keyPath) 这样可以支持wss err = r.Run("127.0.0.1:8888") if err != nil { fmt.Println(err) } }
这样我们的入口就有了~
websocket的模式大概是 onopen onmessage onerror onclose四个callback来覆盖整个通信流程
所以我们来看下简易版本的websockethandler的实现
package main import ( "bytes" "compress/gzip" "encoding/json" "errors" "net/http" "strconv" "time" "util" "github.com/gorilla/websocket" ) var ( ctxHashMap = util.NewConcurrentMap() ) //用来升级http协议到ws协议 type WebSocketHandler struct { wsupgrader websocket.Upgrader } func (wsh *WebSocketHandler) NewWebSocketHandler() { wsh.wsupgrader = websocket.Upgrader{ ReadBufferSize: 4096, WriteBufferSize: 4096, } } func (wsh *WebSocketHandler) onMessage(conn *websocket.Conn, ctx *ConnContext, msg []byte, msgType int) { //处理文本消息 或者 2进制消息 2进制通常是些 gzip的文本 语音或者图片视频之类的一般会用其他云服务不然带宽会爆掉 if msgType == websocket.TextMessage { wsh.processIncomingTextMsg(conn, ctx, msg) } if msgType == websocket.BinaryMessage { } } func (wsh *WebSocketHandler) onOpen(conn *websocket.Conn, r *http.Request) (ctx *ConnContext, err error) { if err := r.ParseForm(); err != nil { return nil, errors.New("参数校验错误") } specialKey := r.FormValue("specialKey") supportGzip := r.FormValue("support_gzip") ctx = &ConnContext{specialKey, supportGzip} //用来标识一个tcp链接 keyString := ctx.AsHashKey() if oldConn, ok := ctxHashMap.Get(keyString); ok { wsh.onClose(oldConn.(*websocket.Conn), ctx) oldConn.(*websocket.Conn).Close() } ctxHashMap.Set(keyString, conn) return ctx, nil } func (wsh *WebSocketHandler) onClose(conn *websocket.Conn, ctx *ConnContext) { logger.Info("client close itself as " + ctx.String()) wsh.closeConnWithCtx(ctx) return } func (wsh *WebSocketHandler) onError(errMsg string) { logger.Error(errMsg) } func (wsh *WebSocketHandler) HandleConn(w http.ResponseWriter, r *http.Request) { wsh.wsupgrader.CheckOrigin = func(r *http.Request) bool { return true } conn, err := wsh.wsupgrader.Upgrade(w, r, nil) if err != nil { logger.Error("Failed to set websocket upgrade: " + err.Error()) return } defer conn.Close() if ctx, err := wsh.onOpen(conn, r); err != nil { logger.Error("Open connection failed " + err.Error() + r.URL.RawQuery) return } else { conn.SetPingHandler(func(message string) error { conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second)) return nil }) for { t, msg, err := conn.ReadMessage() if err != nil { logger.Error("READ ERR FROM " + ctx.String() + " ERR " + err.Error()) wsh.onClose(conn, ctx) return } switch t { case websocket.TextMessage, websocket.BinaryMessage: wsh.onMessage(conn, ctx, msg, t) case websocket.CloseMessage: wsh.onClose(conn, ctx) return case websocket.PingMessage: case websocket.PongMessage: } } } } func (wsh *WebSocketHandler) closeConnWithCtx(ctx *ConnContext) { keyString := ctx.AsHashKey() ctxHashMap.Remove(keyString) return } func (wsh *WebSocketHandler) processIncomingTextMsg(conn *websocket.Conn, ctx *ConnContext, msg []byte) { logger.Debug("CLIENT SAID " + string(msg)) sendMessageToAll(msg) } func (wsh *WebSocketHandler) sendMessageToAll(msg []byte]) { var gzMsg bytes.Buffer gzWriter := gzip.NewWriter(&gzMsg) gzWriter.Write(msg) gzWriter.Flush() gzWriter.Close() for key, conn := range ctxHashMap.Items() { if ctx, err := HashKeyAsCtx(key.(string)); err != nil { wsh.onError(err.Error()) } else { if ctx.supportGzip == "1" { err = conn.(*websocket.Conn).WriteMessage(websocket.BinaryMessage, gzMsg.Bytes()) logger.Debug("send binary msg to " + ctx.String()) } else { err = conn.(*websocket.Conn).WriteMessage(websocket.TextMessage, []byte(msg)) logger.Debug("send text msg to " + ctx.String()) } if err != nil { wsh.onClose(conn.(*websocket.Conn), ctx) conn.(*websocket.Conn).Close() wsh.onError("WRITE ERR TO " + key.(string) + " ERR:" + err.Error()) } } } }
因为删了一些线上代码的敏感信息 所以未必编译的过,不过差不多一个意思,主要看气质
里面的一个莫名其妙的叫做ctx的东西出现了很多次其实是connectionContext的缩写,一般链接形如ws://ip:port/?param=value¶m1=value1之类的形式,当然会加密,所以在onopen的时候会对url做一次基础校验,并且回记录url的一些关键参数标记,以用来确认消息到底要发送给谁
一个简单connContext实现如下
// connContext.go package main import ( "errors" "strings" "util" ) type ConnContext struct { specialKey string supportGzip string } func HashKeyAsCtx(hashKey string) (*ConnContext,error){ values := strings.Split(hashKey,":") if(len(values)!=2){ return nil,errors.New("艾玛 key不对: "+hashKey) }else{ return &ConnContext{values[0],values[1]},nil } } func (ctx *ConnContext) AsHashKey() string{ return strings.Join([]string{ctx.specialKey, ctx.supportGzip},":") } func (ctx * ConnContext) String () string{ return util.NewStringBuilder("specialkey: ",ctx.specialkey, " gzip ",ctx.supportGzip).String() }
以上 一个简易的websocket server 就这样完成了 可喜可贺
有事儿寻这儿