[ go-zero消息中间件[消息队列,定时任务] ]

发表评论
175 views

前置条件

安装中间价 1. kafka+zookeeper消息队列 2. Beanstalks(要求至少2个)+redis定时任务

消息队列api服务目录结构

:.
│  table.api
│  table.go
│
├─etc
│      table-api.yaml
│
└─internal
    ├─config
    │      config.go
    │
    ├─handler
    │      routes.go
    │      tablehandler.go
    │
    ├─logic
    │      tablelogic.go
    │
    ├─mqs
    │      dqs.go
    │      mqs.go
    │      paymentSuccess.go
    │
    ├─svc
    │      servicecontext.go
    │
    └─types
            types.go

消息队列服务端目录结构

配置文件table-api.yaml

添加kafka(消息队列), Beanstalks(延时队列), redis(去重)连接

Name: table-api
Host: 0.0.0.0
Port: 8887

KqConsumerConf:
  Name: kqConsumer
  Brokers:
    - 47.120.55.99:9092
  Group: kqConsumer
  Topic: tables-name
  Offset: first
  Consumers: 8
  Processors: 8

DqConf:
  Beanstalks:
    - Endpoint: 47.120.55.99:11300
      Tube: tube1
    - Endpoint: 47.120.55.99:11301
      Tube: tube2
  Redis:
    Host: 47.120.55.99:6379
    Pass: chenhaifeng
    Type: node

修改table.go文件(api服务入口文件)

修改后把原来添加logic接口改为添加mqs任务

package main

import (
    "context"
    "flag"

    "go-weixin/app/mqueue/table/internal/config"
    "go-weixin/app/mqueue/table/internal/mqs"
    "go-weixin/app/mqueue/table/internal/svc"

    "github.com/zeromicro/go-zero/core/conf"
    "github.com/zeromicro/go-zero/core/service"
    "github.com/zeromicro/go-zero/rest"
)

var configFile = flag.String("f", "etc/table-api.yaml", "the config file")

func main() {
    flag.Parse()

    var c config.Config
    conf.MustLoad(*configFile, &c)

    server := rest.MustNewServer(c.RestConf)
    defer server.Stop()

    svcCtx := svc.NewServiceContext(c)
    ctx := context.Background()
    serviceGroup := service.NewServiceGroup()
    defer serviceGroup.Stop()

    for _, mq := range mqs.Consumers(c, ctx, svcCtx) {
        serviceGroup.Add(mq)
    }

    serviceGroup.Start()
}

mqs文件夹下mqs.go文件(用来添加:消息队列,定时任务,队列的服务端)

package mqs

import (
    "context"
    "go-weixin/app/mqueue/table/internal/config"
    "go-weixin/app/mqueue/table/internal/svc"

    // "github.com/zeromicro/go-queue/dq"
    "github.com/zeromicro/go-queue/kq"
    "github.com/zeromicro/go-zero/core/service"
)

func Consumers(c config.Config, ctx context.Context, svcContext *svc.ServiceContext) []service.Service {

    return []service.Service{
        //Listening for changes in consumption flow status
        kq.MustNewQueue(c.KqConsumerConf, NewMyKQConsume(ctx, svcContext)),
        //.....
        // dq.NewConsumer(c.DqConf),
        NewDqConsumer(ctx, svcContext),
    }

}

一. 消息队列服务添加

服务端

mqs文件夹下的paymentsuccess.go文件

package mqs

import (
    "context"
    "go-weixin/app/mqueue/table/internal/svc"

    "github.com/zeromicro/go-zero/core/logx"
)

type MyKQTask struct {
    ctx    context.Context
    svcCtx *svc.ServiceContext
}

func NewMyKQConsume(ctx context.Context, svcCtx *svc.ServiceContext) *MyKQTask {
    return &MyKQTask{
        ctx:    ctx,
        svcCtx: svcCtx,
    }
}

// 消费者
func (l *MyKQTask) Consume(key, val string) error {
    logx.Infof("执行异步任务 key:%s , val :%s", key, val)
    return nil
}

客户端(在需要发布定时任务的api服务里配置)

  1. etc/yaml配置文件
KqPusherConf:
  Brokers:
    - 47.120.55.99:9092
  Topic: tables-name
  1. config.go
type Config struct {
    ...
    KqPusherConf struct {
        Brokers []string
        Topic   string
    }
    ...
}
  1. svc.go文件
package svc

import (
    "go-weixin/app/usercenter/userapi/internal/config"
    "go-weixin/app/usercenter/userrpc/userrpcclient"
    "go-weixin/common/apiserver"

    "github.com/zeromicro/go-queue/dq"
    "github.com/zeromicro/go-queue/kq"
    "github.com/zeromicro/go-zero/zrpc"
)

type ServiceContext struct {
    ...
    KqClient       *kq.Pusher
    ...
}

func NewServiceContext(c config.Config) *ServiceContext {
    return &ServiceContext{
        ...
        KqClient:       kq.NewPusher(c.KqPusherConf.Brokers, c.KqPusherConf.Topic),
        ...
}
  1. 在logic.go中使用
func (l *TableAddLogic) TableAdd(req *types.TableAddReq) (resp *types.TableAddResp, err error) {
    data := "chenhaifeng"
    if err = l.svcCtx.KqClient.Push(data); err != nil {
        logx.Errorf("KqPusherClient 插入失败, err :%v", err)
        return nil, errors.New("Kq插入失败")
    }
    ...
}

二. 定时任务

服务端

mqs文件夹下的dqs.go文件

package mqs

import (
    "context"
    "go-weixin/app/mqueue/table/internal/svc"

    "github.com/zeromicro/go-zero/core/logx"
)

type MyDqTask struct {
    ctx    context.Context
    svcCtx *svc.ServiceContext
}

func NewDqConsumer(ctx context.Context, svcCtx *svc.ServiceContext) *MyDqTask {
    return &MyDqTask{
        ctx:    ctx,
        svcCtx: svcCtx,
    }
}

func (l *MyDqTask) Start() {
    logx.Info("延时队列启动")
    // 消费者
    l.svcCtx.DqConsumer.Consume(func(body []byte) {
        logx.Infof("执行延时任务: %s \n", string(body))
    })
}

func (l *MyDqTask) Stop() {
    logx.Info("消息队列停止, 假消息")
}

客户端(在需要发布定时任务的api服务里配置)

  1. etc/yaml配置文件
DqConf:
  Beanstalks:
    - Endpoint: 47.120.55.99:11300
      Tube: tube1
    - Endpoint: 47.120.55.99:11301
      Tube: tube2
  Redis:
    Host: 47.120.55.99:6379
    Pass: chenhaifeng
    Type: node
  1. config.go
type Config struct {
    ...
    DqConf    dq.DqConf
    ...
}
  1. svc.go文件
package svc

import (
    "go-weixin/app/usercenter/userapi/internal/config"
    "go-weixin/app/usercenter/userrpc/userrpcclient"
    "go-weixin/common/apiserver"

    "github.com/zeromicro/go-queue/dq"
    "github.com/zeromicro/go-queue/kq"
    "github.com/zeromicro/go-zero/zrpc"
)

type ServiceContext struct {
    ...
    DqPusherClient dq.Producer
    ...
}

func NewServiceContext(c config.Config) *ServiceContext {
    return &ServiceContext{
        ...
        DqPusherClient: dq.NewProducer(c.DqConf.Beanstalks)},
        ...
}
  1. 在logic.go中使用
func (l *TableAddLogic) TableAdd(req *types.TableAddReq) (resp *types.TableAddResp, err error) {
    ...
    role := new(model.RoleModel)
    deplayResp, err := l.svcCtx.DqPusherClient.Delay([]byte(role.TableName()), 60*time.Second)
    if err != nil {
        logx.Infof("延时任务异常: %v", err)
        return nil, err
    }
    logx.Infof("延时任务: %s", deplayResp)

    power := new(model.PowerModel)
    atResp, err := l.svcCtx.DqPusherClient.At([]byte(power.TableName()), time.Now())
    if err != nil {
        logx.Infof("定时任务异常: %v", err)
        return nil, err
    }
    logx.Infof("定时任务: %s", atResp)

    ...
}