99 lines
2.0 KiB
Go
99 lines
2.0 KiB
Go
package task
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"github.com/hibiken/asynq"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// AsynqWorker Asynq Worker
|
|
type AsynqWorker struct {
|
|
server *asynq.Server
|
|
mux *asynq.ServeMux
|
|
taskHandler *ArticleTaskHandler
|
|
logger *zap.Logger
|
|
}
|
|
|
|
// NewAsynqWorker 创建 Asynq Worker
|
|
func NewAsynqWorker(
|
|
redisAddr string,
|
|
taskHandler *ArticleTaskHandler,
|
|
logger *zap.Logger,
|
|
) *AsynqWorker {
|
|
server := asynq.NewServer(
|
|
asynq.RedisClientOpt{Addr: redisAddr},
|
|
asynq.Config{
|
|
Concurrency: 10, // 并发数
|
|
Queues: map[string]int{
|
|
"critical": 6,
|
|
"default": 3,
|
|
"low": 1,
|
|
},
|
|
Logger: NewAsynqLogger(logger),
|
|
},
|
|
)
|
|
|
|
mux := asynq.NewServeMux()
|
|
|
|
return &AsynqWorker{
|
|
server: server,
|
|
mux: mux,
|
|
taskHandler: taskHandler,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
// RegisterHandlers 注册任务处理器
|
|
func (w *AsynqWorker) RegisterHandlers() {
|
|
// 注册文章定时发布任务处理器
|
|
w.mux.HandleFunc(TaskTypeArticlePublish, w.taskHandler.HandleArticlePublish)
|
|
|
|
w.logger.Info("任务处理器注册完成")
|
|
}
|
|
|
|
// Start 启动 Worker
|
|
func (w *AsynqWorker) Start() error {
|
|
w.RegisterHandlers()
|
|
|
|
w.logger.Info("启动 Asynq Worker")
|
|
return w.server.Run(w.mux)
|
|
}
|
|
|
|
// Stop 停止 Worker
|
|
func (w *AsynqWorker) Stop() {
|
|
w.logger.Info("停止 Asynq Worker")
|
|
w.server.Stop()
|
|
w.server.Shutdown()
|
|
}
|
|
|
|
// AsynqLogger Asynq 日志适配器
|
|
type AsynqLogger struct {
|
|
logger *zap.Logger
|
|
}
|
|
|
|
// NewAsynqLogger 创建 Asynq 日志适配器
|
|
func NewAsynqLogger(logger *zap.Logger) *AsynqLogger {
|
|
return &AsynqLogger{logger: logger}
|
|
}
|
|
|
|
func (l *AsynqLogger) Debug(args ...interface{}) {
|
|
l.logger.Debug(fmt.Sprint(args...))
|
|
}
|
|
|
|
func (l *AsynqLogger) Info(args ...interface{}) {
|
|
l.logger.Info(fmt.Sprint(args...))
|
|
}
|
|
|
|
func (l *AsynqLogger) Warn(args ...interface{}) {
|
|
l.logger.Warn(fmt.Sprint(args...))
|
|
}
|
|
|
|
func (l *AsynqLogger) Error(args ...interface{}) {
|
|
l.logger.Error(fmt.Sprint(args...))
|
|
}
|
|
|
|
func (l *AsynqLogger) Fatal(args ...interface{}) {
|
|
l.logger.Fatal(fmt.Sprint(args...))
|
|
}
|