推送服務實作
#基本原則
##server 啟動以後會註冊兩個Handler。 websocketHandler 用於提供瀏覽器端發送 Upgrade 請求並升級為 WebSocket 連線。 pushHandler 用於提供外部推送端發送推送資料的請求。 瀏覽器首先連接 websocketHandler (預設位址為 ws://ip:port/ws)升級請求為 WebSocket 連接,當連接建立之後需要發送註冊資訊進行註冊。這裡註冊資訊中包含一個 token 資訊。 server 會對提供的token 進行驗證並取得到對應的userId(通常來說,一個userId 可能同時關聯許多token),並保存維護好token, userId 和conn(連線)之間的關係。 推送端發送推送資料的請求到 pushHandler(預設位址為 ws://ip:port/push),請求中包含了 userId 欄位和 message 欄位。 server 會根據 userId 取得到所有此時連接到該 server 的 conn,然後將 message 一一進行推送。 由於推送服務的即時性,推送的資料並沒有也不需要進行快取。程式碼詳解
我在這裡會稍微講述一下程式碼的基本構成,也順便說說Go 語言中一些常用的寫法和模式(本人也是從其他語言轉向Go 語言,畢竟Go 語言也相當年輕。所以有建議的話,敬請提出。)。 由於Go語言的發明人和一些主要維護者大都來自於 C/C 語言,所以 Go 語言的程式碼也更偏向 C/C 系。
先來看看 Server 的結構:// Server defines parameters for running websocket server. type Server struct { // Address for server to listen on Addr string // Path for websocket request, default "/ws". WSPath string // Path for push message, default "/push". PushPath string // Upgrader is for upgrade connection to websocket connection using // "github.com/gorilla/websocket". // // If Upgrader is nil, default upgrader will be used. Default upgrader is // set ReadBufferSize and WriteBufferSize to 1024, and CheckOrigin always // returns true. Upgrader *websocket.Upgrader // Check token if it's valid and return userID. If token is valid, userID // must be returned and ok should be true. Otherwise ok should be false. AuthToken func(token string) (userID string, ok bool) // Authorize push request. Message will be sent if it returns true, // otherwise the request will be discarded. Default nil and push request // will always be accepted. PushAuth func(r *http.Request) bool wh *websocketHandler ph *pushHandler }這裡說一下 Upgrader *websocket.Upgrader,這是 gorilla/websocket 套件的對象,它用來升級 HTTP 要求。 如果一個結構體參數太多,通常不建議直接初始化,而是使用它提供的 New 方法。這裡是:
// NewServer creates a new Server.func NewServer(addr string) *Server { return &Server{ Addr: addr, WSPath: serverDefaultWSPath, PushPath: serverDefaultPushPath, } }這也是 Go 語言對外提供初始化方法的常見用法。 然後 Server 使用 ListenAndServe 方法啟動並監聽埠,與 http 套件的使用類似:
// ListenAndServe listens on the TCP network address and handle websocket // request. func (s *Server) ListenAndServe() error { b := &binder{ userID2EventConnMap: make(map[string]*[]eventConn), connID2UserIDMap: make(map[string]string), } // websocket request handler wh := websocketHandler{ upgrader: defaultUpgrader, binder: b, } if s.Upgrader != nil { wh.upgrader = s.Upgrader } if s.AuthToken != nil { wh.calcUserIDFunc = s.AuthToken } s.wh = &wh http.Handle(s.WSPath, s.wh) // push request handler ph := pushHandler{ binder: b, } if s.PushAuth != nil { ph.authFunc = s.PushAuth } s.ph = &ph http.Handle(s.PushPath, s.ph) return http.ListenAndServe(s.Addr, nil) }這裡我們產生了兩個 Handler,分別為 websocketHandler 和 pushHandler。 websocketHandler 負責與瀏覽器建立連線並傳輸數據,而 pushHandler 則處理推播端的請求。 可以看到,這裡兩個 Handler 都封裝了一個 binder 物件。這個 binder 用來維護 token <-> userID <-> Conn 的關係:
// binder is defined to store the relation of userID and eventConn type binder struct { mu sync.RWMutex // map stores key: userID and value of related slice of eventConn userID2EventConnMap map[string]*[]eventConn // map stores key: connID and value: userID connID2UserIDMap map[string]string }websocketHandler##具體看一下 websocketHandler 的實作。
// websocketHandler defines to handle websocket upgrade request. type websocketHandler struct { // upgrader is used to upgrade request. upgrader *websocket.Upgrader // binder stores relations about websocket connection and userID. binder *binder // calcUserIDFunc defines to calculate userID by token. The userID will // be equal to token if this function is nil. calcUserIDFunc func(token string) (userID string, ok bool) }很簡單的結構。 websocketHandler 實作了 http.Handler 介面:
// First try to upgrade connection to websocket. If success, connection will // be kept until client send close message or server drop them. func (wh *websocketHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { wsConn, err := wh.upgrader.Upgrade(w, r, nil) if err != nil { return } defer wsConn.Close() // handle Websocket request conn := NewConn(wsConn) conn.AfterReadFunc = func(messageType int, r io.Reader) { var rm RegisterMessage decoder := json.NewDecoder(r) if err := decoder.Decode(&rm); err != nil { return } // calculate userID by token userID := rm.Token if wh.calcUserIDFunc != nil { uID, ok := wh.calcUserIDFunc(rm.Token) if !ok { return } userID = uID } // bind wh.binder.Bind(userID, rm.Event, conn) } conn.BeforeCloseFunc = func() { // unbind wh.binder.Unbind(conn) } conn.Listen() }首先將傳入的 http.Request 轉換為 websocket.Conn,再將其分裝為我們自訂的一個 wserver.Conn(封裝,或說是組合,是Go 語言的典型用法。記住,Go 語言沒有繼承,只有組合)。 接著設定了 Conn 的 AfterReadFunc 和 BeforeCloseFunc 方法,接著啟動了 conn.Listen()。 AfterReadFunc 意思是當 Conn 讀取到資料後,請嘗試驗證並依據 token 計算 userID,且為 bind 註冊綁定。 BeforeCloseFunc 則為 Conn 關閉前進行解綁作業。 pushHandlerpushHandler 則是容易理解。它解析請求然後推送資料:
// Authorize if needed. Then decode the request and push message to each // realted websocket connection. func (s *pushHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { w.WriteHeader(http.StatusMethodNotAllowed) return } // authorize if s.authFunc != nil { if ok := s.authFunc(r); !ok { w.WriteHeader(http.StatusUnauthorized) return } } // read request var pm PushMessage decoder := json.NewDecoder(r.Body) if err := decoder.Decode(&pm); err != nil { w.WriteHeader(http.StatusBadRequest) w.Write([]byte(ErrRequestIllegal.Error())) return } // validate the data if pm.UserID == "" || pm.Event == "" || pm.Message == "" { w.WriteHeader(http.StatusBadRequest) w.Write([]byte(ErrRequestIllegal.Error())) return } cnt, err := s.push(pm.UserID, pm.Event, pm.Message) if err != nil { w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(err.Error())) return } result := strings.NewReader(fmt.Sprintf("message sent to %d clients", cnt)) io.Copy(w, result) } Conn Conn (此处指 wserver.Conn) 为 websocket.Conn 的包装。 // Conn wraps websocket.Conn with Conn. It defines to listen and read // data from Conn. type Conn struct { Conn *websocket.Conn AfterReadFunc func(messageType int, r io.Reader) BeforeCloseFunc func() once sync.Once id string stopCh chan struct{} }最主要的方法為 Listen():###
// Listen listens for receive data from websocket connection. It blocks // until websocket connection is closed. func (c *Conn) Listen() { c.Conn.SetCloseHandler(func(code int, text string) error { if c.BeforeCloseFunc != nil { c.BeforeCloseFunc() } if err := c.Close(); err != nil { log.Println(err) } message := websocket.FormatCloseMessage(code, "") c.Conn.WriteControl(websocket.CloseMessage, message, time.Now().Add(time.Second)) return nil }) // Keeps reading from Conn util get error. ReadLoop: for { select { case <-c.stopCh: break ReadLoop default: messageType, r, err := c.Conn.NextReader() if err != nil { // TODO: handle read error maybe break ReadLoop } if c.AfterReadFunc != nil { c.AfterReadFunc(messageType, r) } } } }
主要設定了 websocket 連線關閉時的處理和不停地讀取資料。
推薦:g
olang教學以上是用Go語言寫一個簡單的WebSocket推送服務的詳細內容。更多資訊請關注PHP中文網其他相關文章!