這篇文章主要講解了“構(gòu)建一個即時消息應用之什么是實時消息”,文中的講解內(nèi)容簡單清晰,易于學習與理解,下面請大家跟著小編的思路慢慢深入,一起來研究和學習“構(gòu)建一個即時消息應用之什么是實時消息”吧!
成都創(chuàng)新互聯(lián)專注于企業(yè)網(wǎng)絡營銷推廣、網(wǎng)站重做改版、貢嘎網(wǎng)站定制設計、自適應品牌網(wǎng)站建設、H5響應式網(wǎng)站、成都商城網(wǎng)站開發(fā)、集團公司官網(wǎng)建設、成都外貿(mào)網(wǎng)站建設、高端網(wǎng)站制作、響應式網(wǎng)頁設計等建站業(yè)務,價格優(yōu)惠性價比高,為貢嘎等各大城市提供網(wǎng)站開發(fā)制作服務。
在 HTTP 部分之前,讓我們先編寫一個映射 ,讓所有客戶端都監(jiān)聽消息。 像這樣全局初始化:
type MessageClient struct { Messages chan Message UserID string } var messageClients sync.Map
還記得在 上一篇文章 中,當我們創(chuàng)建這條消息時,我們留下了一個 “TODO” 注釋。在那里,我們將使用這個函數(shù)來調(diào)度一個 goroutine。
go messageCreated(message)
把這行代碼插入到我們留注釋的位置。
func messageCreated(message Message) error { if err := db.QueryRow(` SELECT user_id FROM participants WHERE user_id != $1 and conversation_id = $2 `, message.UserID, message.ConversationID). Scan(&message.ReceiverID); err != nil { return err } go broadcastMessage(message) return nil } func broadcastMessage(message Message) { messageClients.Range(func(key, _ interface{}) bool { client := key.(*MessageClient) if client.UserID == message.ReceiverID { client.Messages <- message } return true }) }
該函數(shù)查詢接收者 ID(其他參與者 ID),并將消息發(fā)送給所有客戶端。
讓我們轉(zhuǎn)到 main()
函數(shù)并添加以下路由:
router.HandleFunc("GET", "/api/messages", guard(subscribeToMessages))
此端點處理 /api/messages
上的 GET 請求。請求應該是一個 EventSource 連接。它用一個事件流響應,其中的數(shù)據(jù)是 JSON 格式的。
func subscribeToMessages(w http.ResponseWriter, r *http.Request) { if a := r.Header.Get("Accept"); !strings.Contains(a, "text/event-stream") { http.Error(w, "This endpoint requires an EventSource connection", http.StatusNotAcceptable) return } f, ok := w.(http.Flusher) if !ok { respondError(w, errors.New("streaming unsupported")) return } ctx := r.Context() authUserID := ctx.Value(keyAuthUserID).(string) h := w.Header() h.Set("Cache-Control", "no-cache") h.Set("Connection", "keep-alive") h.Set("Content-Type", "text/event-stream") messages := make(chan Message) defer close(messages) client := &MessageClient{Messages: messages, UserID: authUserID} messageClients.Store(client, nil) defer messageClients.Delete(client) for { select { case <-ctx.Done(): return case message := <-messages: if b, err := json.Marshal(message); err != nil { log.Printf("could not marshall message: %v\n", err) fmt.Fprintf(w, "event: error\ndata: %v\n\n", err) } else { fmt.Fprintf(w, "data: %s\n\n", b) } f.Flush() } } }
首先,它檢查請求頭是否正確,并檢查服務器是否支持流式傳輸。我們創(chuàng)建一個消息通道,用它來構(gòu)建一個客戶端,并將其存儲在客戶端映射中。每當創(chuàng)建新消息時,它都會進入這個通道,因此我們可以通過 for-select
循環(huán)從中讀取。
服務器發(fā)送事件使用以下格式發(fā)送數(shù)據(jù):
data: some data here\n\n
我們以 JSON 格式發(fā)送:
data: {"foo":"bar"}\n\n
我們使用 fmt.Fprintf()
以這種格式寫入響應寫入器,并在循環(huán)的每次迭代中刷新數(shù)據(jù)。
這個循環(huán)會一直運行,直到使用請求上下文關(guān)閉連接為止。我們延遲了通道的關(guān)閉和客戶端的刪除,因此,當循環(huán)結(jié)束時,通道將被關(guān)閉,客戶端不會收到更多的消息。
注意,服務器發(fā)送事件(EventSource)的 JavaScript API 不支持設置自定義請求頭?,所以我們不能設置 Authorization: Bearer
。這就是為什么 guard()
中間件也會從 URL 查詢字符串中讀取令牌的原因。
感謝各位的閱讀,以上就是“構(gòu)建一個即時消息應用之什么是實時消息”的內(nèi)容了,經(jīng)過本文的學習后,相信大家對構(gòu)建一個即時消息應用之什么是實時消息這一問題有了更深刻的體會,具體使用情況還需要大家實踐驗證。這里是創(chuàng)新互聯(lián),小編將為大家推送更多相關(guān)知識點的文章,歡迎關(guān)注!