diff --git a/internal/msggateway/online.go b/internal/msggateway/online.go index b50608f93..9604d563b 100644 --- a/internal/msggateway/online.go +++ b/internal/msggateway/online.go @@ -4,13 +4,16 @@ import ( "context" "crypto/md5" "encoding/binary" + "fmt" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" pbuser "github.com/openimsdk/protocol/user" "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/utils/datautil" "math/rand" + "os" "strconv" + "sync/atomic" "time" ) @@ -78,8 +81,10 @@ func (ws *WsServer) ChangeOnlineStatus(concurrent int) { } } - opIdCtx := mcontext.SetOperationID(context.Background(), "r"+strconv.FormatUint(rNum, 10)) + var count atomic.Int64 + operationIDPrefix := fmt.Sprintf("p_%d_", os.Getpid()) doRequest := func(req *pbuser.SetUserOnlineStatusReq) { + opIdCtx := mcontext.SetOperationID(context.Background(), operationIDPrefix+strconv.FormatInt(count.Add(1), 10)) ctx, cancel := context.WithTimeout(opIdCtx, time.Second*5) defer cancel() if _, err := ws.userClient.Client.SetUserOnlineStatus(ctx, req); err != nil { @@ -102,7 +107,7 @@ func (ws *WsServer) ChangeOnlineStatus(concurrent int) { case now := <-renewalTicker.C: deadline := now.Add(-cachekey.OnlineExpire / 3) users := ws.clients.GetAllUserStatus(deadline, now) - log.ZDebug(context.Background(), "renewal ticker", "deadline", deadline, "nowtime", now, "num", len(users)) + log.ZDebug(context.Background(), "renewal ticker", "deadline", deadline, "nowtime", now, "users", users) pushUserState(users...) case state := <-ws.clients.UserState(): log.ZDebug(context.Background(), "OnlineCache user online change", "userID", state.UserID, "online", state.Online, "offline", state.Offline) diff --git a/internal/msggateway/user_map.go b/internal/msggateway/user_map.go index bd1f19728..14e7c66ac 100644 --- a/internal/msggateway/user_map.go +++ b/internal/msggateway/user_map.go @@ -1,7 +1,13 @@ package msggateway import ( + "bytes" + "context" + "fmt" + "github.com/openimsdk/protocol/constant" + "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/utils/datautil" + "strconv" "sync" "time" ) @@ -27,6 +33,23 @@ type UserPlatform struct { Clients []*Client } +func (u *UserPlatform) String() string { + buf := bytes.NewBuffer(nil) + buf.WriteString("UserPlatform{Time: ") + buf.WriteString(u.Time.Format(time.DateTime)) + buf.WriteString(", Clients<") + buf.WriteString(strconv.Itoa(len(u.Clients))) + buf.WriteString(">: [") + for i, client := range u.Clients { + if i > 0 { + buf.WriteString(", ") + } + buf.WriteString(fmt.Sprintf("%p %d %s", u.Clients[i], client.PlatformID, constant.PlatformIDToName(client.PlatformID))) + } + buf.WriteString("]}") + return buf.String() +} + func (u *UserPlatform) PlatformIDs() []int32 { if len(u.Clients) == 0 { return nil @@ -117,6 +140,7 @@ func (u *userMap) Get(userID string, platformID int) ([]*Client, bool, bool) { } func (u *userMap) Set(userID string, client *Client) { + log.ZDebug(context.Background(), "userMap Set", "userID", userID, "platformID", client.PlatformID, "platform", constant.PlatformIDToName(client.PlatformID), "pointer", fmt.Sprintf("%p", client)) u.lock.Lock() defer u.lock.Unlock() result, ok := u.data[userID] @@ -135,6 +159,9 @@ func (u *userMap) DeleteClients(userID string, clients []*Client) (isDeleteUser if len(clients) == 0 { return false } + for i, client := range clients { + log.ZDebug(context.Background(), "userMap DeleteClients", "userID", userID, "platformID", client.PlatformID, "platform", constant.PlatformIDToName(client.PlatformID), "count", fmt.Sprintf("%p %d/%d", clients[i], i+1, len(clients))) + } u.lock.Lock() defer u.lock.Unlock() result, ok := u.data[userID] @@ -167,6 +194,7 @@ func (u *userMap) GetAllUserStatus(deadline time.Time, nowtime time.Time) []User defer u.lock.RUnlock() result := make([]UserState, 0, len(u.data)) for userID, userPlatform := range u.data { + log.ZDebug(context.Background(), "userMap GetAllUserStatus", "userID", userID, "platforms", userPlatform.String()) if userPlatform.Time.Before(deadline) { continue }