99 lines
1.9 KiB
Go
99 lines
1.9 KiB
Go
package worker
|
|
|
|
import (
|
|
"91porn-server/common/log"
|
|
"errors"
|
|
"fmt"
|
|
"runtime"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
stop = 0
|
|
running = 1
|
|
)
|
|
|
|
type Worker struct {
|
|
num int // 最大工作协程，限制同时执行任务的协程数量，避免同时执行的任务太多，超过其他服务的承载上限
|
|
ch chan struct{}
|
|
wg *sync.WaitGroup
|
|
state int
|
|
lock *sync.RWMutex
|
|
}
|
|
|
|
func NewWorker(workerNum int) *Worker {
|
|
return &Worker{
|
|
num: workerNum,
|
|
ch: make(chan struct{}, workerNum),
|
|
wg: &sync.WaitGroup{},
|
|
state: running,
|
|
lock: &sync.RWMutex{},
|
|
}
|
|
}
|
|
|
|
func (w *Worker) wait() {
|
|
w.wg.Wait()
|
|
}
|
|
|
|
func (w *Worker) Exec(f func()) (err error) {
|
|
defer func() {
|
|
if e := recover(); e != nil {
|
|
log.Error("worker exec Panic: ", log.Any("e", e))
|
|
}
|
|
}()
|
|
tick := time.NewTicker(time.Second * 1)
|
|
if w.State() != running {
|
|
return
|
|
}
|
|
select {
|
|
case w.ch <- struct{}{}:
|
|
go func() {
|
|
w.wg.Add(1)
|
|
defer func() {
|
|
<-w.ch
|
|
w.wg.Done()
|
|
if r := recover(); r != nil {
|
|
dep := 0
|
|
t := make([]string, 0, 10)
|
|
for i := 1; i < 10; i++ {
|
|
_, file, line, ok := runtime.Caller(i)
|
|
if !ok {
|
|
break
|
|
}
|
|
if strings.Contains(file, "/runtime/") || strings.Contains(file, "/reflect/") {
|
|
continue
|
|
}
|
|
t = append(t, fmt.Sprintf("%s∟%s:%d", strings.Repeat(" ", dep), file, line))
|
|
dep++
|
|
}
|
|
exception := strings.Join(t, "\n")
|
|
log.Error("worker exec func Panic: ", log.Any("Exception", exception))
|
|
}
|
|
}()
|
|
f()
|
|
}()
|
|
case <-tick.C:
|
|
log.Debug("worker exec timeout")
|
|
// 等待执行超时，直接返回
|
|
return errors.New("Exec timeout")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (w *Worker) Stop() {
|
|
w.lock.Lock()
|
|
w.state = stop
|
|
w.lock.Unlock()
|
|
// 等待所有的任务执行完毕再退出
|
|
w.wait()
|
|
return
|
|
}
|
|
|
|
func (w *Worker) State() int {
|
|
w.lock.RLock()
|
|
defer w.lock.RUnlock()
|
|
return w.state
|
|
}
|