用戶連接流程： 客戶端:  ~~~ ***獲取節點 getScript() http://localhost:8090/1/server/get?k=Terry-Mao&p=2 [先根據用戶的key獲取連接節點，然后再根據對應的節點創建長連接（參數p=1 websocket,p=2 tcp）] 創建websocket連接 獲取離線消息 建立心跳任務 初始化完成（等待接收在線消息 or 心跳應答） ~~~ 服務端: ~~~ 建立連接： SubscribeHandle() 監聽websocket請求 addr := ws.Request().RemoteAddr 獲取用戶ip地址 params := ws.Request().URL.Query() 獲取url參數 key、heartbeat、token、version UserChannel.Get(key, true) 在ChannelList上添加一個Channel，并返回這個用戶的 Channel c.AuthToken(key, token) token驗證（默認不驗證） c.AddConn(key, &Connection{Conn: ws, Proto: WebsocketProto, Version: version})創建一個用戶連接， 1.判斷是否超過最大連接數 2.用戶鏈接應答 3.conn.HandleWrite(key) 開啟 goroutine 從 Connection.Buf chan []byte 中讀取消息，推送給用戶 4.c.conn.PushFront(conn) root *Element 上添加一個 *Element 5.ConnStat.IncrAdd() 總連接數+1 for{} 阻塞等待心跳->心跳應答 獲取離線消息： GetOfflineMsg() 獲取離線消息 r.Method != "GET" 判斷請求方式 params := r.URL.Query() 獲取url參數 key、msgId、callback myrpc.MessageRPC.Get() 隨機獲取一個rpc連接 client.Call(myrpc.MessageServiceGetPrivate, args, reply) rpc 到message GetPrivate()方法 UseStorage.GetPrivate(m.Key, m.MsgId) 根據key和msgId從redis or mysql中獲取消息 redis： 1.conn := s.getConn(key) 根據key從連接池獲取一個redis連接（hash算法） 2.redis.Values(conn.Do("ZRANGEBYSCORE", key, fmt.Sprintf("(%d", mid), "+inf", "WITHSCORES")) 返回所有符合條件 mid < msgId <= +inf(最大值)) 的成員及成員的 message 3.redis.Scan(values, &b, &cmid) 遍歷消息 4.json.Unmarshal(b, rm) 消息反序列化 5.rm.Expire < now 判斷消息是否過期 6.s.delCH <- &RedisDelMessage{Key: key, MIds: delMsgs}: 刪除unmarshal失敗的消息和過期消息（clean() 方法） mysql： 1.s.getConn(key) 獲取mysql連接 2.db.Query(getPrivateMsgSQL, key, mid) 查詢skey=key and mid>msgId 的消息 "SELECT mid, ttl, msg FROM private_msg WHERE skey=? AND mid>? ORDER BY mid" 3.for rows.Next() 遍歷消息 4.now > expire 判斷消息是否過期 ~~~ 推送單個私信流程：  ~~~ PushPrivate() 推送單個私信 r.Method != "POST" 判斷是否是post請求 ioutil.ReadAll(r.Body) 讀取請求內容 params := r.URL.Query() 獲取url參數 key、expire node := myrpc.GetComet(key) 根據key獲取comet節點 ？？？[判斷key連接的comet] client := node.Rpc.Get() 隨機獲取一個rpc連接 client.Call(myrpc.CometServicePushPrivate, args, &ret) rpc 調用 comet 推送私信 PushPrivate() UserChannel.New(args.Key) 獲取用戶channel ch.PushMsg(args.Key, m, args.Expire) 推送私信 1.client := myrpc.MessageRPC.Get() 隨機獲取一個RPC連接 2.m.MsgId = id.Get() 生成msgId （時間戳/100） 3.m.GroupId != myrpc.PublicGroupId && expire > 0 判斷是是否需要message保存（私信+過期時間>0） 4.client.Call(myrpc.MessageServiceSavePrivate, args, &ret) rpc 調用 message 模塊，保存消息 SavePrivate() 保存私信 UseStorage.SavePrivate(m.Key, m.Msg, m.MsgId, m.Expire) redis: 1.conn := s.getConn(key) 根據key 通過hash算法從連接池獲取一個連接 2.conn.Send("ZADD", key, mid, m) 操作寫入緩沖區 3.conn.Send("ZREMRANGEBYRANK", key, 0, -1*(Conf.RedisMaxStore+1)) // conn.Send("ZREMRANGEBYRANK", key, 0, -21) 有序集只剩下最后寫入的20個成員 4.conn.Flush() 提交操作 5.conn.Receive() 接受redis應答 mysql: 1.db := s.getConn(key) 根據key 通過hash算法從連接池獲取一個連接 2.b.Exec(savePrivateMsgSQL, key, mid, now.Unix()+int64(expire), []byte(msg), now, now) 存mysql 5. c.writeMsg(key, m) 推送在線消息 ~~~ 批量推送私信流程：  ~~~ PushMultiPrivate() 批量推送私信 r.Method != "POST" 判斷是否是post請求 ioutil.ReadAll(r.Body) 讀取請求內容 parseMultiPrivate(bodyBytes) 獲取 keys 、 message 和 ret（錯誤碼） 根據key獲取node，通過node獲取rpc 鏈接 rpc -> comet PushPrivates() 向多個key推送私信 UserChannel.New(key) 根據key獲取一個channel 和 ChannelBucket 并保存到 bucketMap 遍歷 bucketMap 開啟 goroutine 存儲消息 和 推送消息 1.獲取請求消息 2.根據請求消息解析出keys和message 3.根據key匹配節點，存儲到 map[node]keys 4.rpc 調用用節點的 PushPrivates() 方法 5.判斷 key 所在的bucketChannel 返回 map[*ChannelBucket]*batchChannel 6.遍歷 ChannelBucket 給keys 發消息，并rpc 調用 Message模塊 存儲消息 ~~~ writeMsg： 1.調用每一個 用戶 Element 元素 2.將消息寫入Connection.Buf chan []byte 3.HandleWrite() 消費 Buf chan 將消息推給用戶 MsgId: 通過當前時間的納秒時間戳(除) / 100 獲得 存儲時隨存儲到redis中，用于在用戶獲取離線消息時，匹配 MsgId 大于 用戶所傳參的值