liuxiaobo/fox
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修复阻塞导致无法处理退出信号

Browse Source
This commit is contained in:
liuxiaobo 2025-05-31 09:35:55 +08:00
parent 9a6a73cae5
commit cb572a5d8f
2 changed files with 15 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
safeChan/safeChan.go
View File

@ -15,6 +15,7 @@ type SafeChan[T any] struct {
once sync.Once once sync.Once
} }
// size为0会创建无缓冲队列，读chan时，没有数据会阻塞到有数据可读
func NewSafeChan[T any](size int) *SafeChan[T] { func NewSafeChan[T any](size int) *SafeChan[T] {
ch := &SafeChan[T]{} ch := &SafeChan[T]{}
ch.ctx, ch.cancel = context.WithCancel(context.Background()) ch.ctx, ch.cancel = context.WithCancel(context.Background())
@ -40,6 +41,7 @@ func (s *SafeChan[T]) Size() int {
} }
func (s *SafeChan[T]) Reader() <-chan T { func (s *SafeChan[T]) Reader() <-chan T {
//log.Debug("reader channel")
return s.ch return s.ch
} }

14
service/baseService.go
View File

@ -57,6 +57,7 @@ func (s *BaseService) Write(msg []byte) error {
return s.msg.Write(msg) return s.msg.Write(msg)
} }
// 执行一次回调
func (s *BaseService) RunOnce(cb func()) { func (s *BaseService) RunOnce(cb func()) {
select { select {
case <-s.readyStop.Done(): case <-s.readyStop.Done():
@ -67,6 +68,7 @@ func (s *BaseService) RunOnce(cb func()) {
} }
} }
// 执行一次回调并等待返回值
func (s *BaseService) RunWait(cb func() (retValue any)) (retValue any, err error) { func (s *BaseService) RunWait(cb func() (retValue any)) (retValue any, err error) {
select { select {
case <-s.readyStop.Done(): case <-s.readyStop.Done():
@ -113,7 +115,7 @@ func (s *BaseService) WaitStop() {
func (s *BaseService) NotifyStop() { func (s *BaseService) NotifyStop() {
s.readyStopFunc() s.readyStopFunc()
// log.Debug(fmt.Sprintf("notify %v service readyStop", s.name)) //log.Debug(s.Log("notify %v service readyStop", s.name))
} }
//func (s *BaseService) allChanEmpty() bool { //func (s *BaseService) allChanEmpty() bool {
@ -126,8 +128,10 @@ func (s *BaseService) NotifyStop() {
func (s *BaseService) canStop() bool { func (s *BaseService) canStop() bool {
select { select {
case <-s.readyStop.Done(): case <-s.readyStop.Done():
//log.Error(s.Log("want readyStop"))
return true return true
default: default:
//log.Error(s.Log("can not Stop"))
return false return false
} }
} }
@ -135,6 +139,7 @@ func (s *BaseService) canStop() bool {
func (s *BaseService) run() { func (s *BaseService) run() {
for { for {
if s.onFunc.CanStop() && s.canStop() { if s.onFunc.CanStop() && s.canStop() {
//log.Error(s.Log("stop service"))
s.msg.Close() s.msg.Close()
s.job.Close() s.job.Close()
s.Timer.CancelAllTimer() s.Timer.CancelAllTimer()
@ -144,19 +149,26 @@ func (s *BaseService) run() {
s.waitStopFunc() s.waitStopFunc()
break break
} }
// select无default会阻塞在case的chan中直到有可读chan
select { select {
case msg, ok := <-s.msg.Reader(): case msg, ok := <-s.msg.Reader():
//log.Debug(s.Log("msg reader"))
if ok { if ok {
_ = s.onFunc.OnMessage(msg) _ = s.onFunc.OnMessage(msg)
} }
case cb, ok := <-s.job.Reader(): case cb, ok := <-s.job.Reader():
//log.Debug(s.Log("job reader"))
if ok && cb != nil { if ok && cb != nil {
cb() cb()
} }
case t, ok := <-s.Timer.Reader(): case t, ok := <-s.Timer.Reader():
//log.Debug(s.Log("timer reader"))
if ok && t != nil && t.Func != nil { if ok && t != nil && t.Func != nil {
t.Func() t.Func()
} }
default:
// 休眠50微秒，避免cpu占用过高
time.Sleep(50 * time.Microsecond)
} }
} }
} }