From 6a09618e9df8c84f5497b4403e94d518277ca5eb Mon Sep 17 00:00:00 2001 From: zjwshisb <361340272@qq.com> Date: Tue, 21 Jan 2025 14:24:53 +0800 Subject: [PATCH] feat: optimize code --- internal/logic/chat/admin_manager.go | 2 +- internal/logic/chat/client.go | 5 +--- internal/logic/chat/manager.go | 35 ++++++++-------------------- internal/logic/chat/user_manager.go | 2 +- 4 files changed, 13 insertions(+), 31 deletions(-) diff --git a/internal/logic/chat/admin_manager.go b/internal/logic/chat/admin_manager.go index b33985d..4154d3d 100644 --- a/internal/logic/chat/admin_manager.go +++ b/internal/logic/chat/admin_manager.go @@ -20,7 +20,7 @@ import ( func newAdminManager(cluster bool) *adminManager { adminM = &adminManager{ - newManager(10, 100, time.Minute, cluster, consts.WsTypeAdmin), + newManager(10, time.Minute, cluster, consts.WsTypeAdmin), } adminM.on(eventRegister, adminM.onRegister) adminM.on(eventUnRegister, adminM.onUnRegister) diff --git a/internal/logic/chat/client.go b/internal/logic/chat/client.go index 0823717..be23c53 100644 --- a/internal/logic/chat/client.go +++ b/internal/logic/chat/client.go @@ -211,10 +211,7 @@ func (c *client) readMsg() { } msg.CustomerId = c.getCustomerId() msg.ReceivedAt = gtime.Now() - c.manager.receiveMessage(&chatConnMessage{ - Msg: msg, - Conn: c, - }) + c.manager.handleMessage(c, msg) c.lastActive = gtime.Now() } } diff --git a/internal/logic/chat/manager.go b/internal/logic/chat/manager.go index 638aa31..8dd9d53 100644 --- a/internal/logic/chat/manager.go +++ b/internal/logic/chat/manager.go @@ -48,8 +48,7 @@ type connManager interface { run() ping() SendAction(act *api.ChatAction, conn ...iWsConn) - receiveMessage(cm *chatConnMessage) - handleReceiveMessage() + handleMessage(conn iWsConn, msg *model.CustomerChatMessage) noticeRead(ctx context.Context, customerId uint, uid uint, msgIds []uint, forceLocal ...bool) error } @@ -58,10 +57,9 @@ type eventArg struct { msg *model.CustomerChatMessage } -func newManager(shareCount uint, msgCount int, pingDuration time.Duration, cluster bool, types string) *manager { +func newManager(shareCount uint, pingDuration time.Duration, cluster bool, types string) *manager { return &manager{ shardCount: shareCount, - connMessages: make(chan *chatConnMessage, msgCount), events: nil, pingDuration: pingDuration, cluster: cluster, @@ -72,7 +70,6 @@ func newManager(shareCount uint, msgCount int, pingDuration time.Duration, clust type manager struct { shardCount uint // 分组数量, 默认 10 shard []*shard // 分组切片 - connMessages chan *chatConnMessage // 接受从conn所读取消息的chan events map[string][]eventHandle // 事件 pingDuration time.Duration // default to 10 seconds cluster bool @@ -143,25 +140,14 @@ func (m *manager) getSpread(customerId uint) *shard { return m.shard[m.getMod(customerId)] } -// ReceiveMessage 接受消息 -func (m *manager) receiveMessage(cm *chatConnMessage) { - m.connMessages <- cm -} - -// 从conn接受消息并处理 -func (m *manager) handleReceiveMessage() { - for { - payload := <-m.connMessages - go func() { - ctx := gctx.New() - err := m.trigger(ctx, eventMessage, eventArg{ - conn: payload.Conn, - msg: payload.Msg, - }) - if err != nil { - log.Errorf(ctx, "%+v", err) - } - }() +func (m *manager) handleMessage(conn iWsConn, msg *model.CustomerChatMessage) { + ctx := gctx.New() + err := m.trigger(ctx, eventMessage, eventArg{ + conn: conn, + msg: msg, + }) + if err != nil { + log.Errorf(ctx, "%+v", err) } } @@ -430,6 +416,5 @@ func (m *manager) run() { mutex: &sync.RWMutex{}, } } - go m.handleReceiveMessage() go m.ping() } diff --git a/internal/logic/chat/user_manager.go b/internal/logic/chat/user_manager.go index 1d0977e..09012fc 100644 --- a/internal/logic/chat/user_manager.go +++ b/internal/logic/chat/user_manager.go @@ -22,7 +22,7 @@ import ( func newUserManager(cluster bool) *userManager { userM = &userManager{ - newManager(10, 100, time.Minute, cluster, consts.WsTypeUser), + newManager(10, time.Minute, cluster, consts.WsTypeUser), } userM.on(eventRegister, userM.onRegister) userM.on(eventUnRegister, userM.onUnRegister)