初始化

apps/queue/etc/queue-dev.yaml

@@ -0,0 +1,23 @@
+Name: zero-ppanel-queue
+Mode: dev
+
+Log:
+  Mode: console
+  Encoding: plain
+  Level: debug
+
+Telemetry:
+  Name: zero-ppanel-queue
+  Endpoint: 127.0.0.1:4318
+  Sampler: 1.0
+  Batcher: otlphttp
+
+MySQL:
+  DataSource: "root:password@tcp(127.0.0.1:3306)/ppanel?charset=utf8mb4&parseTime=true"
+
+Redis:
+  Host: 127.0.0.1:6379
+  Type: node
+
+Asynq:
+  Addr: 127.0.0.1:6379

apps/queue/etc/queue-prod.yaml

@@ -0,0 +1,28 @@
+Name: zero-ppanel-queue
+Mode: pro
+
+Log:
+  Mode: file
+  Encoding: json
+  Level: info
+  Path: /var/log/zero-ppanel/queue
+  KeepDays: 15
+  Rotation: daily
+
+Telemetry:
+  Name: zero-ppanel-queue
+  Endpoint: http://jaeger:4318/v1/traces
+  Sampler: 0.1
+  Batcher: otlphttp
+
+MySQL:
+  DataSource: "${MYSQL_DSN}"
+
+Redis:
+  Host: "${REDIS_HOST}"
+  Type: node
+  Pass: "${REDIS_PASS}"
+
+Asynq:
+  Addr: "${REDIS_HOST}"
+  Pass: "${REDIS_PASS}"

apps/queue/etc/queue-test.yaml

@@ -0,0 +1,25 @@
+Name: zero-ppanel-queue
+Mode: test
+
+Log:
+  Mode: file
+  Encoding: json
+  Level: info
+  Path: logs/queue
+  KeepDays: 7
+
+Telemetry:
+  Name: zero-ppanel-queue
+  Endpoint: http://jaeger:4318/v1/traces
+  Sampler: 0.5
+  Batcher: otlphttp
+
+MySQL:
+  DataSource: "root:password@tcp(mysql:3306)/ppanel?charset=utf8mb4&parseTime=true"
+
+Redis:
+  Host: redis:6379
+  Type: node
+
+Asynq:
+  Addr: redis:6379

apps/queue/internal/config/config.go

@@ -0,0 +1,30 @@
+package config
+
+import (
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/trace"
+)
+
+type Config struct {
+	Name string
+	Mode string `json:",default=pro"`
+
+	Log logx.LogConf
+
+	Telemetry trace.Config `json:",optional"`
+
+	MySQL struct {
+		DataSource string
+	}
+
+	Redis struct {
+		Host string
+		Type string
+		Pass string `json:",optional"`
+	}
+
+	Asynq struct {
+		Addr string
+		Pass string `json:",optional"`
+	}
+}

apps/queue/internal/handler/routes.go

@@ -0,0 +1,15 @@
+package handler
+
+import (
+	"github.com/hibiken/asynq"
+
+	"github.com/zero-ppanel/zero-ppanel/apps/queue/internal/svc"
+)
+
+func RegisterHandlers(mux *asynq.ServeMux, svcCtx *svc.ServiceContext) {
+	// TODO: Register asynq task handlers
+	// mux.HandleFunc("task:send_email", logic.NewSendEmailLogic(svcCtx).Handle)
+	// mux.HandleFunc("task:send_sms", logic.NewSendSmsLogic(svcCtx).Handle)
+	// mux.HandleFunc("task:order_timeout", logic.NewOrderTimeoutLogic(svcCtx).Handle)
+	// mux.HandleFunc("task:telegram_notify", logic.NewTelegramNotifyLogic(svcCtx).Handle)
+}

apps/queue/internal/svc/servicecontext.go

@@ -0,0 +1,15 @@
+package svc
+
+import (
+	"github.com/zero-ppanel/zero-ppanel/apps/queue/internal/config"
+)
+
+type ServiceContext struct {
+	Config config.Config
+}
+
+func NewServiceContext(c config.Config) *ServiceContext {
+	return &ServiceContext{
+		Config: c,
+	}
+}

apps/queue/internal/types/types.go

@@ -0,0 +1,9 @@
+package types
+
+const (
+	TaskSendEmail      = "task:send_email"
+	TaskSendSms        = "task:send_sms"
+	TaskOrderTimeout   = "task:order_timeout"
+	TaskTelegramNotify = "task:telegram_notify"
+	TaskSendMessage    = "task:send_message"
+)

apps/queue/queue.go

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+
+	"github.com/hibiken/asynq"
+	"github.com/zero-ppanel/zero-ppanel/apps/queue/internal/config"
+	"github.com/zero-ppanel/zero-ppanel/apps/queue/internal/handler"
+	"github.com/zero-ppanel/zero-ppanel/apps/queue/internal/svc"
+	"github.com/zeromicro/go-zero/core/conf"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/trace"
+)
+
+var configFile = flag.String("f", "etc/queue-dev.yaml", "the config file")
+
+func main() {
+	flag.Parse()
+
+	var c config.Config
+	conf.MustLoad(*configFile, &c)
+
+	logx.MustSetup(c.Log)
+	if c.Telemetry.Name != "" {
+		trace.StartAgent(c.Telemetry)
+	}
+
+	svcCtx := svc.NewServiceContext(c)
+
+	server := asynq.NewServer(
+		asynq.RedisClientOpt{Addr: c.Asynq.Addr, Password: c.Asynq.Pass},
+		asynq.Config{
+			Concurrency: 20,
+			IsFailure: func(err error) bool {
+				logx.Errorf("asynq task error: %v", err)
+				return true
+			},
+		},
+	)
+
+	mux := asynq.NewServeMux()
+	handler.RegisterHandlers(mux, svcCtx)
+
+	fmt.Println("Starting zero-ppanel-queue worker...")
+	if err := server.Run(mux); err != nil {
+		logx.Must(err)
+	}
+}