Ma/origin
1
0
Fork 0
mirror of https://github.com/duanhf2012/origin.git synced 2026-03-12 11:07:53 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

优化并发模块

Browse Source
This commit is contained in:
duanhf2012
2023-02-22 15:13:51 +08:00
parent 8cc1b1afcb
commit 4783d05e75
2 changed files with 6 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
concurrent/concurrent.go
View File

@@ -69,6 +69,10 @@ func (c *Concurrent) AsyncDoByQueue(queueId int64, fn func(), cb func(err error)
return return
} }
if queueId != 0 {
queueId = queueId % maxTaskQueueSessionId+1
}
select { select {
case c.tasks <- task{queueId, fn, cb}: case c.tasks <- task{queueId, fn, cb}:
} }

3
concurrent/dispatch.go
View File

@@ -13,6 +13,7 @@ import (
) )
var idleTimeout = 2 * time.Second var idleTimeout = 2 * time.Second
const maxTaskQueueSessionId = 10000
type dispatch struct { type dispatch struct {
minConcurrentNum int32 minConcurrentNum int32
@@ -35,7 +36,7 @@ func (d *dispatch) open(minGoroutineNum int32, maxGoroutineNum int32, tasks chan
d.minConcurrentNum = minGoroutineNum d.minConcurrentNum = minGoroutineNum
d.maxConcurrentNum = maxGoroutineNum d.maxConcurrentNum = maxGoroutineNum
d.tasks = tasks d.tasks = tasks
d.mapTaskQueueSession = make(map[int64]*queue.Deque[task], 1024) d.mapTaskQueueSession = make(map[int64]*queue.Deque[task], maxTaskQueueSessionId)
d.workerQueue = make(chan task) d.workerQueue = make(chan task)
d.cbChannel = cbChannel d.cbChannel = cbChannel
d.queueIdChannel = make(chan int64, cap(tasks)) d.queueIdChannel = make(chan int64, cap(tasks))