91 lines
1.9 KiB
Go
91 lines
1.9 KiB
Go
|
/*
|
|||
|
|
* Copyright (C) THL A29 Limited, a Tencent company. All rights reserved.
|
|||
|
|
*
|
|||
|
|
* SPDX-License-Identifier: Apache-2.0
|
|||
|
|
*
|
|||
|
|
*/
|
|||
|
|
|
|||
|
|
package lockfree
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"fmt"
|
|||
|
|
"runtime"
|
|||
|
|
"sync/atomic"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
// consumer 消费者，这个消费者只会有一个g操作，这样处理的好处是可以不涉及并发操作，其内部不会涉及到任何锁
|
|||
|
|
// 对于实际的并发操作由该g进行分配
|
|||
|
|
type consumer[T any] struct {
|
|||
|
|
status int32 // 运行状态
|
|||
|
|
rbuf *ringBuffer[T]
|
|||
|
|
seqer *sequencer
|
|||
|
|
blocks blockStrategy
|
|||
|
|
hdl EventHandler[T]
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func newConsumer[T any](rbuf *ringBuffer[T], hdl EventHandler[T], sequer *sequencer, blocks blockStrategy) *consumer[T] {
|
|||
|
|
return &consumer[T]{
|
|||
|
|
rbuf: rbuf,
|
|||
|
|
seqer: sequer,
|
|||
|
|
hdl: hdl,
|
|||
|
|
blocks: blocks,
|
|||
|
|
status: READY,
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (c *consumer[T]) start() error {
|
|||
|
|
if atomic.CompareAndSwapInt32(&c.status, READY, RUNNING) {
|
|||
|
|
go c.handle()
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
return fmt.Errorf(StartErrorFormat, "Consumer")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (c *consumer[T]) handle() {
|
|||
|
|
// 判断是否可以获取到
|
|||
|
|
rc := c.seqer.nextRead()
|
|||
|
|
for {
|
|||
|
|
if c.closed() {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
var i = 0
|
|||
|
|
for {
|
|||
|
|
if c.closed() {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
// 看下读取位置的seq是否OK
|
|||
|
|
if v, p, exist := c.rbuf.contains(rc - 1); exist {
|
|||
|
|
rc = c.seqer.readIncrement()
|
|||
|
|
c.hdl.OnEvent(v)
|
|||
|
|
i = 0
|
|||
|
|
break
|
|||
|
|
} else {
|
|||
|
|
if i < spin {
|
|||
|
|
procyield(30)
|
|||
|
|
} else if i < spin+passiveSpin {
|
|||
|
|
runtime.Gosched()
|
|||
|
|
} else {
|
|||
|
|
c.blocks.block(p, rc)
|
|||
|
|
i = 0
|
|||
|
|
}
|
|||
|
|
i++
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (c *consumer[T]) close() error {
|
|||
|
|
if atomic.CompareAndSwapInt32(&c.status, RUNNING, READY) {
|
|||
|
|
// 防止阻塞无法释放
|
|||
|
|
c.blocks.release()
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
return fmt.Errorf(CloseErrorFormat, "Consumer")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// closed 判断是否已关闭
|
|||
|
|
// 将直接判断调整为原子操作，解决data race问题
|
|||
|
|
func (c *consumer[T]) closed() bool {
|
|||
|
|
return atomic.LoadInt32(&c.status) == READY
|
|||
|
|
}
|