goim 中的 data flow 數據流轉及思考

時間 2019-11-07

原文原文鏈接

goim 文章系列(共5篇):html

goim 架構與定製

從goim定製, 淺談 golang 的 interface 解耦合與gRPC

goim中的 bilibili/discovery (eureka)基本概念及應用

goim 的 data flow 數據流

goim的業務集成(分享會小結與QA)

有個 slack 頻道, 很多朋友在交流 goim , 歡迎加入slack #goimgit

[簡述] goim.io 是很是成功的 IM (Instance Message) 即時消息平臺 , 本文介紹 goim 中的數據定義與 data flow 數據流轉github

1. goim 中的 data flow 數據流轉

1.1 架構中的數據流轉

看圖golang

數據流轉web

http 接口向 logic 發送數據
logic 從 redis 中獲取會話數據, 以 protobuf 序列化數據, 發送到 MQ
job 從 MQ 中訂閱數據, 取出 im 發送數據中的 server / room 向指定 server 的 comet 發送數據
comet 接收 job 分發的數據後, 存入指定 channel 的 ring buffer , 再轉爲 tcp/websocket 數據包, 發送到指定 channel 的客戶端

1.2 簡化後的數據流轉細節

上示意圖標註了 goim 中的關鍵數據結構:redis

標註了 im 發送數據構成, 注意, 這個數據結構是被logic 以 protobuf 序列化後發到 MQ , 並在 job 中反序列化後, 分發到 comet
這裏的會話信息, 主要是 mid --> server 與 room-->server 的對應關係, 存在 redis 中
comet 中的 im 信息, 由 job 從 MQ 中反序列化後, 取出 server / room / keys( 一到多個key , 對應 channel ) 發送到指定 comet server
comet 以 tcp / websocket 封裝數據包, 發送給終端用戶, 終端解包後顯示

2. goim 中的數據定義

2.1. logic 發送 im 信息

發佈 im 信息定義( 在 protobuf 中的定義)json

message PushMsg {
    enum Type {
        PUSH = 0;
        ROOM = 1;
        BROADCAST = 2;
    }
    Type type = 1;
    int32 operation = 2;
    int32 speed = 3;
    string server = 4;
    string room = 5;
    repeated string keys = 6;
    bytes msg = 7;
}
複製代碼

2.2 會話數據

當 tcp client 或 websocket client 鏈接 comet server 時, comet 以 gRPC 向 logic 進行內部通信, 生成會話數據, 存在 redis 中, 具體細節不展開, 看代碼api

當 http client 向 logic 發送 im 消息時, logic 向 redis 查詢會話數據, 對於已經存在的 room--> server / mid ( memberID) --> server 即發送消息到 MQ , 該部分代碼比較清楚, 也再也不加說明緩存

2.3. tcp / websocket 數據包定義

推送 im 信息, 對象名稱爲 proto, 在 protobuf 中定義bash

message Proto {
    int32 ver = 1 [(gogoproto.jsontag) = "ver"];
    int32 op = 2 [(gogoproto.jsontag) = "op"];
    int32 seq = 3 [(gogoproto.jsontag) = "seq"];
    bytes body = 4 [(gogoproto.jsontag) = "body"];
}
複製代碼

protobuf 文件 github.com/Terry-Mao/g… 中第12行

tcp / websocket 數據包組包/折包操做在 /api/comet/grpc/protocol.go

由上圖可見, goim 在 tcp /websocket 數據包的數據包定義, 與 go 中 proto 定義, 多了, 數據包總長度 / 包頭長度兩個字段

3. comet 中的處理

簡化數據流轉, 從發送端數據到接收端數據, 能夠看到, serverID / roomID / channel ( 用 mid 或 key 來指示) 的主要做用做爲分流/分發用, 在最後推送數據包中, 就不在包含這三個字段了.

同時, comet 中使用了 ring buffer 來緩存一個 channel 送達的多條信息並推送到終端, 這裏, 並無看到對推送下發的信息做更多處理.

看代碼, 補充細節

// Channel used by message pusher send msg to write goroutine.
type Channel struct {
	c        *conf.CometConfig
	Room     *Room
	CliProto Ring
	signal   chan *grpc.Proto
	Writer   xbufio.Writer
	Reader   xbufio.Reader
	Next     *Channel
	Prev     *Channel

	Mid      int64   // ######### memberID 
	Key      string
	IP       string
	watchOps map[int32]struct{}

	mutex sync.RWMutex
}
複製代碼

這裏:

mid 就是 memberID , 當前 channel ( 用戶端與 comet 的長鏈接) 是哪一個用戶鏈接上的該長鏈接使用 key 做爲長鏈接的會話標識, 換個方式說, key 也就標定了一個 im 信息要發給哪一個/哪幾個在線長鏈接對端的用戶
key 就是長鏈接的會話ID, 能夠這麼理解, 就算是 sessionID 吧
watchOps 是一個map 映射表, 其中的 int32 是房間號. map 多個房間號, map 結構是用來查詢房間號是否在 map 中存在或不存在. watchOps 是當前長鏈接用戶用來監聽當前客戶端接收哪一個房間的 im 消息推送, 換個方式說, 一個 goim 終端能夠接收多個房間發送來的 im 消息
watchOps 初始化是在 tcp / websocket 客戶端進行首次鏈接時處理的, 細節看代碼.

從 logic 自 http 的 post 請求中, 獲取發佈 im 信息後, 序列化發到 MQ, 在 job 中拆包反序列化, 再組包, 這一步驟對性能是否有影響, 需發測試數據來定位, 但我的感受, 這幾回拆包組包, 有點重複.