fusen/fusenapi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

11

Browse Source
This commit is contained in:
laodaming 2023-08-28 10:00:18 +08:00
parent 1120dec011
commit d130f1390f
1 changed files with 24 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
server/websocket/internal/logic/commonnotifylogic.go
View File

@ -37,7 +37,7 @@ type commonConnectionNotFoundDataCacheChanItem struct {
} }
// 放入缓冲队列 // 放入缓冲队列
func (l *CommonNotifyLogic) pushCommonCache(data commonConnectionNotFoundDataCacheChanItem) { func (l *CommonNotifyLogic) pushCommonNotifyCache(data commonConnectionNotFoundDataCacheChanItem) {
select { select {
case commonConnectionNotFoundDataCacheChan <- data: case commonConnectionNotFoundDataCacheChan <- data:
return return
@ -47,22 +47,20 @@ func (l *CommonNotifyLogic) pushCommonCache(data commonConnectionNotFoundDataCac
} }
// 取出元素 // 取出元素
func (l *CommonNotifyLogic) popCommonCache() (data commonConnectionNotFoundDataCacheChanItem) { func (l *CommonNotifyLogic) popCommonNotifyCache() (data commonConnectionNotFoundDataCacheChanItem) {
return <-commonConnectionNotFoundDataCacheChan return <-commonConnectionNotFoundDataCacheChan
} }
// 保证处理消息就一个循环在执行 // 保证处理消息就一个循环在执行
var consumeCommonCacheData sync.Once var consumeCommonCacheData sync.Once
// 消费公共通知未处理的消息 // 消费公共通知未处理的消息(目前是轮巡方式，待优化)
func (l *CommonNotifyLogic) consumeCommonCacheData() { func (l *CommonNotifyLogic) consumeCommonCacheData() {
//单例 //单例
consumeCommonCacheData.Do(func() { consumeCommonCacheData.Do(func() {
tick := time.Tick(time.Millisecond * 200)
for { for {
select { time.Sleep(time.Millisecond * 200)
case <-tick: //200毫秒触发一次 info := l.popCommonNotifyCache()
info := l.popCommonCache()
//查询websocket连接 //查询websocket连接
value, ok := mapConnPool.Load(info.data.Wid) value, ok := mapConnPool.Load(info.data.Wid)
//没有连接 //没有连接
@ -70,7 +68,7 @@ func (l *CommonNotifyLogic) consumeCommonCacheData() {
info.retryTimes-- info.retryTimes--
//大于0，则放回队列 //大于0，则放回队列
if info.retryTimes > 0 { if info.retryTimes > 0 {
l.pushCommonCache(info) l.pushCommonNotifyCache(info)
continue continue
} }
//否则直接丢弃消息 //否则直接丢弃消息
@ -85,8 +83,6 @@ func (l *CommonNotifyLogic) consumeCommonCacheData() {
//发送 //发送
ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, info.data.Data)) ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, info.data.Data))
} }
}
}) })
} }
@ -105,7 +101,7 @@ func (l *CommonNotifyLogic) CommonNotify(req *types.CommonNotifyReq, userinfo *a
value, ok := mapConnPool.Load(req.Wid) value, ok := mapConnPool.Load(req.Wid)
if !ok { if !ok {
//没找到连接就放到公共缓冲队列 //没找到连接就放到公共缓冲队列
go l.pushCommonCache(commonConnectionNotFoundDataCacheChanItem{ go l.pushCommonNotifyCache(commonConnectionNotFoundDataCacheChanItem{
retryTimes: 20, //重试20次 retryTimes: 20, //重试20次
data: types.CommonNotifyReq{ data: types.CommonNotifyReq{
Wid: req.Wid, Wid: req.Wid,