-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
other: v2 vivid 实现优先级邮箱、 pulse 采用优先级邮箱
- Loading branch information
1 parent
9f92fd4
commit 96d6da9
Showing
13 changed files
with
320 additions
and
43 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| package vivid | ||
|
|
||
| import ( | ||
| "container/heap" | ||
| "sync" | ||
| ) | ||
|
|
||
| const ( | ||
| priorityStateNone = priorityState(iota) // 未启动状态 | ||
| priorityStateRunning // 运行中状态 | ||
| priorityStateStopping // 停止中状态 | ||
| priorityStateStopped // 已停止状态 | ||
| ) | ||
|
|
||
| const ( | ||
| PriorityStopModeInstantly = PriorityStopMode(iota) // 立刻停止消息队列，新消息将不再接收，缓冲区内未处理的消息将被丢弃 | ||
| PriorityStopModeGraceful // 优雅停止消息队列，新消息将不再接收，等待未处理的消息处理完毕后再停止 | ||
| PriorityStopModeDrain // 新消息将继续被接收，等待消息队列处理完毕且没有新消息后再停止 | ||
| ) | ||
|
|
||
| type ( | ||
| priorityState = int32 // 状态 | ||
| PriorityStopMode = int8 // Priority 消息队列的停止模式，目前支持 PriorityStopModeInstantly、 PriorityStopModeGraceful、 PriorityStopModeDrain | ||
| ) | ||
|
|
||
| func NewPriority(handler func(message MessageContext), opts ...*PriorityOptions) *Priority { | ||
| p := &Priority{ | ||
| opts: NewPriorityOptions().Apply(opts...), | ||
| status: priorityStateNone, | ||
| cond: sync.NewCond(&sync.Mutex{}), | ||
| closed: make(chan struct{}), | ||
| handler: handler, | ||
| } | ||
| p.buffer = make(priorityHeap, 0, p.opts.BufferSize) | ||
| return p | ||
| } | ||
|
|
||
| type Priority struct { | ||
| opts *PriorityOptions // 配置 | ||
| status priorityState // 队列状态 | ||
| cond *sync.Cond // 消息队列条件变量 | ||
| buffer priorityHeap // 消息缓冲区 | ||
| closed chan struct{} // 关闭信号 | ||
| handler func(message MessageContext) // 消息处理函数 | ||
| } | ||
|
|
||
| func (p *Priority) Start() { | ||
| p.cond.L.Lock() | ||
| if p.status != priorityStateNone { | ||
| p.cond.L.Unlock() | ||
| return | ||
| } | ||
| p.status = priorityStateRunning | ||
| p.cond.L.Unlock() | ||
|
|
||
| p.closed = make(chan struct{}) | ||
| go func(p *Priority) { | ||
| defer func(p *Priority) { | ||
| close(p.closed) | ||
| if err := recover(); err != nil { | ||
| panic(err) | ||
| } | ||
| }(p) | ||
|
|
||
| for { | ||
| p.cond.L.Lock() | ||
| if p.buffer.Len() == 0 { | ||
| if p.status == priorityStateStopping { | ||
| p.status = priorityStateStopped | ||
| p.cond.L.Unlock() | ||
| break | ||
| } | ||
| p.cond.Wait() | ||
| if p.buffer.Len() == 0 { | ||
| p.cond.L.Unlock() | ||
| continue | ||
| } | ||
| } | ||
| msg := heap.Pop(&p.buffer).(MessageContext) | ||
| p.cond.L.Unlock() | ||
|
|
||
| p.handler(msg) | ||
| } | ||
| }(p) | ||
| } | ||
|
|
||
| func (p *Priority) Stop() { | ||
| p.cond.L.Lock() | ||
| if p.status != priorityStateRunning { | ||
| p.cond.L.Unlock() | ||
| return | ||
| } | ||
| p.status = priorityStateStopping | ||
| p.cond.L.Unlock() | ||
| p.cond.Signal() | ||
| } | ||
|
|
||
| func (p *Priority) Enqueue(message MessageContext) bool { | ||
| p.cond.L.Lock() | ||
| if p.status != priorityStateRunning { | ||
| p.cond.L.Unlock() | ||
| return false | ||
| } | ||
| heap.Push(&p.buffer, message) | ||
| p.cond.L.Unlock() | ||
| p.cond.Signal() | ||
| return true | ||
| } | ||
|
|
||
| func (p *Priority) reset() { | ||
| p.cond.L.Lock() | ||
| p.buffer = p.buffer[:0] | ||
| p.cond.L.Unlock() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| package vivid | ||
|
|
||
| import ( | ||
| "github.com/kercylan98/minotaur/toolkit/pools" | ||
| ) | ||
|
|
||
| func NewPriorityFactory(handler func(message MessageContext), opts ...*PriorityOptions) MailboxFactory { | ||
| var pool = pools.NewObjectPool[Priority](func() *Priority { | ||
| return NewPriority(handler, opts...) | ||
| }, func(data *Priority) { | ||
| data.reset() | ||
| }) | ||
|
|
||
| return &PriorityFactory{pool: pool} | ||
| } | ||
|
|
||
| type PriorityFactory struct { | ||
| pool *pools.ObjectPool[*Priority] | ||
| } | ||
|
|
||
| func (P *PriorityFactory) Get() Mailbox { | ||
| return P.pool.Get() | ||
| } | ||
|
|
||
| func (P *PriorityFactory) Put(mailbox Mailbox) { | ||
| priority, ok := mailbox.(*Priority) | ||
| if !ok { | ||
| return | ||
| } | ||
| P.pool.Release(priority) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| package vivid | ||
|
|
||
| type priorityHeap []MessageContext | ||
|
|
||
| func (h *priorityHeap) Len() int { | ||
| return len(*h) | ||
| } | ||
|
|
||
| func (h *priorityHeap) Less(i, j int) bool { | ||
| return (*h)[i].GetPriority() < (*h)[j].GetPriority() | ||
| } | ||
|
|
||
| func (h *priorityHeap) Swap(i, j int) { | ||
| (*h)[i], (*h)[j] = (*h)[j], (*h)[i] | ||
| } | ||
|
|
||
| func (h *priorityHeap) Push(x any) { | ||
| *h = append(*h, x.(MessageContext)) | ||
| } | ||
|
|
||
| func (h *priorityHeap) Pop() any { | ||
| old := *h | ||
| n := len(old) | ||
| x := old[n-1] | ||
| *h = old[0 : n-1] | ||
| return x | ||
| } |
Oops, something went wrong.