blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-06 04:15:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

perf: reduce register client latency (#1312)

Browse Source 
Signed-off-by: rfyiamcool <rfyiamcool@163.com>
This commit is contained in:
fengyun.rui 2023-11-02 12:11:19 +08:00 committed by GitHub
parent cc1f7739e1
commit 1664579cf7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 29 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
internal/msggateway/n_ws_server.go
View File

@ -74,8 +74,8 @@ type WsServer struct {
kickHandlerChan chan *kickHandler kickHandlerChan chan *kickHandler
clients *UserMap clients *UserMap
clientPool sync.Pool clientPool sync.Pool
onlineUserNum int64 onlineUserNum atomic.Int64
onlineUserConnNum int64 onlineUserConnNum atomic.Int64
handshakeTimeout time.Duration handshakeTimeout time.Duration
hubServer *Server hubServer *Server
validate *validator.Validate validate *validator.Validate
@ -220,8 +220,8 @@ func (ws *WsServer) registerClient(client *Client) {
if !userOK { if !userOK {
ws.clients.Set(client.UserID, client) ws.clients.Set(client.UserID, client)
log.ZDebug(client.ctx, "user not exist", "userID", client.UserID, "platformID", client.PlatformID) log.ZDebug(client.ctx, "user not exist", "userID", client.UserID, "platformID", client.PlatformID)
atomic.AddInt64(&ws.onlineUserNum, 1) ws.onlineUserNum.Add(1)
atomic.AddInt64(&ws.onlineUserConnNum, 1) ws.onlineUserConnNum.Add(1)
} else { } else {
i := &kickHandler{ i := &kickHandler{
clientOK: clientOK, clientOK: clientOK,
@ -234,22 +234,35 @@ func (ws *WsServer) registerClient(client *Client) {
ws.clients.Set(client.UserID, client) ws.clients.Set(client.UserID, client)
// 已经有同平台的连接存在 // 已经有同平台的连接存在
log.ZInfo(client.ctx, "repeat login", "userID", client.UserID, "platformID", client.PlatformID, "old remote addr", getRemoteAdders(oldClients)) log.ZInfo(client.ctx, "repeat login", "userID", client.UserID, "platformID", client.PlatformID, "old remote addr", getRemoteAdders(oldClients))
atomic.AddInt64(&ws.onlineUserConnNum, 1) ws.onlineUserConnNum.Add(1)
} else { } else {
ws.clients.Set(client.UserID, client) ws.clients.Set(client.UserID, client)
ws.onlineUserConnNum.Add(1)
}
}
atomic.AddInt64(&ws.onlineUserConnNum, 1) wg := sync.WaitGroup{}
} wg.Add(1)
} go func() {
ws.sendUserOnlineInfoToOtherNode(client.ctx, client) defer wg.Done()
_ = ws.sendUserOnlineInfoToOtherNode(client.ctx, client)
}()
wg.Add(1)
go func() {
defer wg.Done()
ws.SetUserOnlineStatus(client.ctx, client, constant.Online) ws.SetUserOnlineStatus(client.ctx, client, constant.Online)
}()
wg.Wait()
log.ZInfo( log.ZInfo(
client.ctx, client.ctx,
"user online", "user online",
"online user Num", "online user Num",
ws.onlineUserNum, ws.onlineUserNum.Load(),
"online user conn Num", "online user conn Num",
ws.onlineUserConnNum, ws.onlineUserConnNum.Load(),
) )
} }
@ -282,7 +295,7 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
if clientOK { if clientOK {
isDeleteUser := ws.clients.deleteClients(newClient.UserID, oldClients) isDeleteUser := ws.clients.deleteClients(newClient.UserID, oldClients)
if isDeleteUser { if isDeleteUser {
atomic.AddInt64(&ws.onlineUserNum, -1) ws.onlineUserNum.Add(-1)
} }
for _, c := range oldClients { for _, c := range oldClients {
err := c.KickOnlineMessage() err := c.KickOnlineMessage()
@ -350,18 +363,18 @@ func (ws *WsServer) unregisterClient(client *Client) {
defer ws.clientPool.Put(client) defer ws.clientPool.Put(client)
isDeleteUser := ws.clients.delete(client.UserID, client.ctx.GetRemoteAddr()) isDeleteUser := ws.clients.delete(client.UserID, client.ctx.GetRemoteAddr())
if isDeleteUser { if isDeleteUser {
atomic.AddInt64(&ws.onlineUserNum, -1) ws.onlineUserNum.Add(-1)
} }
atomic.AddInt64(&ws.onlineUserConnNum, -1) ws.onlineUserConnNum.Add(-1)
ws.SetUserOnlineStatus(client.ctx, client, constant.Offline) ws.SetUserOnlineStatus(client.ctx, client, constant.Offline)
log.ZInfo(client.ctx, "user offline", "close reason", client.closedErr, "online user Num", ws.onlineUserNum, "online user conn Num", log.ZInfo(client.ctx, "user offline", "close reason", client.closedErr, "online user Num", ws.onlineUserNum, "online user conn Num",
ws.onlineUserConnNum, ws.onlineUserConnNum.Load(),
) )
} }
func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) { func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
connContext := newContext(w, r) connContext := newContext(w, r)
if ws.onlineUserConnNum >= ws.wsMaxConnNum { if ws.onlineUserConnNum.Load() >= ws.wsMaxConnNum {
httpError(connContext, errs.ErrConnOverMaxNumLimit) httpError(connContext, errs.ErrConnOverMaxNumLimit)
return return
} }