微服务从代码到k8s部署应有尽有系列（八、各种队列）

var configFile = flag.String("f", "etc/order.yaml", "Specify the config file")
func main() {
    flag.Parse()
    var c config.Config
    conf.MustLoad(*configFile, &c)
    // log, prometheus, trace, metricsUrl
    if err := c.SetUp(); err != nil {
        panic(err)
    }
    serviceGroup := service.NewServiceGroup()
    defer serviceGroup.Stop()
    for _, mq := range listen.Mqs(c) {
        serviceGroup.Add(mq)
    }
    serviceGroup.Start()
}

// Service is the interface that groups Start and Stop methods.
Service interface {
    Starter // Start
    Stopper // Stop
}

// 返回所有消费者
func Mqs(c config.Config) []service.Service {
    svcContext := svc.NewServiceContext(c)
    ctx := context.Background()
    var services []service.Service
    // kq ：消息队列.
    services = append(services, KqMqs(c, ctx, svcContext)...)
    // asynq：延迟队列、定时任务
    services = append(services, AsynqMqs(c, ctx, svcContext)...)
    // other mq ....
    return services
}

// asynq
// 定时任务、延迟任务
func AsynqMqs(c config.Config, ctx context.Context, svcContext *svc.ServiceContext) []service.Service {
   return []service.Service{
      // 监听延迟队列
      deferMq.NewAsynqTask(ctx, svcContext),
      // 监听定时任务
   }
}

// kq
// 消息队列
func KqMqs(c config.Config, ctx context.Context, svcContext *svc.ServiceContext) []service.Service {
    return []service.Service{
        // 监听消费流水状态变更
        kq.MustNewQueue(c.PaymentUpdateStatusConf, kqMq.NewPaymentUpdateStatusMq(ctx, svcContext)),
        // .....
    }
}

// 监听关闭订单
type AsynqTask struct {
   ctx    context.Context
   svcCtx *svc.ServiceContext
}
func NewAsynqTask(ctx context.Context, svcCtx *svc.ServiceContext) *AsynqTask {
   return &AsynqTask{
      ctx:    ctx,
      svcCtx: svcCtx,
   }
}
func (l *AsynqTask) Start() {
   fmt.Println("AsynqTask start ")
   srv := asynq.NewServer(
      asynq.RedisClientOpt{Addr: l.svcCtx.Config.Redis.Host, Password: l.svcCtx.Config.Redis.Pass},
      asynq.Config{
         Concurrency: 10,
         Queues: map[string]int{
            "critical": 6,
            "default":  3,
            "low":      1,
         },
      },
   )
   mux := asynq.NewServeMux()
   // 关闭民宿订单任务
   mux.HandleFunc(asynqmq.TypeHomestayOrderCloseDelivery, l.closeHomestayOrderStateMqHandler)
   if err := srv.Run(mux); err != nil {
      log.Fatalf("could not run server: %v", err)
   }
}
func (l *AsynqTask) Stop() {
   fmt.Println("AsynqTask stop")
}

package deferMq
import (
    "context"
    "encoding/json"
    "looklook/app/order/cmd/rpc/order"
    "looklook/app/order/model"
    "looklook/common/asynqmq"
    "looklook/common/xerr"
    "github.com/hibiken/asynq"
    "github.com/pkg/errors"
)
func (l *AsynqTask) closeHomestayOrderStateMqHandler(ctx context.Context, t *asynq.Task) error {
    var p asynqmq.HomestayOrderCloseTaskPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return errors.Wrapf(xerr.NewErrMsg("解析asynq task payload err"), "closeHomestayOrderStateMqHandler payload err:%v, payLoad:%+v", err, t.Payload())
    }
    resp, err := l.svcCtx.OrderRpc.HomestayOrderDetail(ctx, &order.HomestayOrderDetailReq{
        Sn: p.Sn,
    })
    if err != nil || resp.HomestayOrder == nil {
        return errors.Wrapf(xerr.NewErrMsg("获取订单失败"), "closeHomestayOrderStateMqHandler 获取订单失败 or 订单不存在 err:%v, sn:%s ,HomestayOrder : %+v", err, p.Sn, resp.HomestayOrder)
    }
    if resp.HomestayOrder.TradeState == model.HomestayOrderTradeStateWaitPay {
        _, err := l.svcCtx.OrderRpc.UpdateHomestayOrderTradeState(ctx, &order.UpdateHomestayOrderTradeStateReq{
            Sn:         p.Sn,
            TradeState: model.HomestayOrderTradeStateCancel,
        })
        if err != nil {
            return errors.Wrapf(xerr.NewErrMsg("关闭订单失败"), "closeHomestayOrderStateMqHandler 关闭订单失败  err:%v, sn:%s ", err, p.Sn)
        }
    }
    return nil
}

func KqMqs(c config.Config, ctx context.Context, svcContext *svc.ServiceContext) []service.Service {
    return []service.Service{
        // 监听消费流水状态变更
        kq.MustNewQueue(c.PaymentUpdateStatusConf, kqMq.NewPaymentUpdateStatusMq(ctx, svcContext)),
        // .....
    }
}

func (l *PaymentUpdateStatusMq) Consume(_, val string) error {
    fmt.Printf(" PaymentUpdateStatusMq Consume val : %s \n", val)
    // 解析数据
    var message kqueue.ThirdPaymentUpdatePayStatusNotifyMessage
    if err := json.Unmarshal([]byte(val), &message); err != nil {
        logx.WithContext(l.ctx).Error("PaymentUpdateStatusMq->Consume Unmarshal err : %v , val : %s", err, val)
        return err
    }
    // 执行业务..
    if err := l.execService(message); err != nil {
        logx.WithContext(l.ctx).Error("PaymentUpdateStatusMq->execService  err : %v , val : %s , message:%+v", err, val, message)
        return err
    }
    return nil
}

version: '3'
services:
  #asynqmon asynq延迟队列、定时队列的webui
  asynqmon:
    image: hibiken/asynqmon:latest
    container_name: asynqmon_asynq
    ports:
      - 8980:8080
    command:
      - '--redis-addr=redis:6379'
      - '--redis-password=G62m50oigInC30sf'
    restart: always
    networks:
      - asynqtest_net
    depends_on:
      - redis
  
  #redis容器
  redis:
    image: redis:6.2.5
    container_name: redis_asynq
    ports:
      - 63779:6379
    environment:
      # 时区上海
      TZ: Asia/Shanghai
    volumes:
      # 数据文件
      - ./data/redis/data:/data:rw
    command: "redis-server --requirepass G62m50oigInC30sf  --appendonly yes"
    privileged: true
    restart: always
    networks:
      - asynqtest_net
networks:
  asynqtest_net:
    driver: bridge
    ipam:
      config:
        - subnet: 172.22.0.0/16

package main
import (
    "asynqtest/tpl"
    "encoding/json"
    "log"
    "github.com/hibiken/asynq"
)
const redisAddr = "127.0.0.1:63779"
const redisPwd = "G62m50oigInC30sf"
func main() {
    // 周期性任务
    scheduler := asynq.NewScheduler(
        asynq.RedisClientOpt{
            Addr:     redisAddr,
            Password: redisPwd,
        }, nil)
    payload, err := json.Marshal(tpl.EmailPayload{Email: "546630576@qq.com", Content: "发邮件呀"})
    if err != nil {
        log.Fatal(err)
    }
    task := asynq.NewTask(tpl.EMAIL_TPL, payload)
    // 每隔1分钟同步一次
    entryID, err := scheduler.Register("*/1 * * * *", task)
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("registered an entry: %q\n", entryID)
    if err := scheduler.Run(); err != nil {
        log.Fatal(err)
    }
}

package main
import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "asynqtest/tpl"
    "github.com/hibiken/asynq"
)
func main() {
    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: "127.0.0.1:63779", Password: "G62m50oigInC30sf"},
        asynq.Config{
            Concurrency: 10,
            Queues: map[string]int{
                "critical": 6,
                "default":  3,
                "low":      1,
            },
        },
    )
    mux := asynq.NewServeMux()
    // 关闭民宿订单任务
    mux.HandleFunc(tpl.EMAIL_TPL, emailMqHandler)
    if err := srv.Run(mux); err != nil {
        log.Fatalf("could not run server: %v", err)
    }
}
func emailMqHandler(ctx context.Context, t *asynq.Task) error {
    var p tpl.EmailPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("emailMqHandler err:%+v", err)
    }
    fmt.Printf("p : %+v \n", p)
    return nil
}

package tpl
const EMAIL_TPL = "schedule:email"
type EmailPayload struct {
    Email   string
    Content string
}