Queue

[!TIP] This document is machine-translated by Google. If you find grammatical and semantic errors, and the document description is not clear, please PR

In the development of daily tasks, we will have many asynchronous, batch, timing, and delayed tasks to be processed. There is go-queue in go-zero. It is recommended to use go-queue for processing. Go-queue itself is also developed based on go-zero. There are two modes

  • dq : Depends on beanstalkd, distributed, can be stored, delayed, timing settings, shutdown and restart can be re-executed, messages will not be lost, very simple to use, redis setnx is used in go-queue to ensure that each message is only consumed once, usage scenarios Mainly used for daily tasks.

  • kq: Depends on Kafka, so I won’t introduce more about it, the famous Kafka, the usage scenario is mainly to do message queue

    We mainly talk about dq. The use of kq is also the same, but it depends on the bottom layer. If you haven’t used beanstalkd, you can google it first. It’s still very easy to use.

etc/job.yaml : Configuration file

  1.   Name: job
  3.   Log:
  4.     ServiceName: job
  5.     Level: info
  7.   # dq depends on Beanstalks, redis, Beanstalks configuration, redis configuration
  8.   DqConf:
  9.     Beanstalks:
  10.       - Endpoint: 127.0.0.1:7771
  11.         Tube: tube1
  12.       - Endpoint: 127.0.0.1:7772
  13.         Tube: tube2
  14.     Redis:
  15.       Host: 127.0.0.1:6379
  16.       Type: node

Internal/config/config.go: Parse dq corresponding etc/*.yaml configuration

  1.   /**
  2.   * @Description Configuration file
  3.   * @Author Mikael
  4.   * @Email 13247629622@163.com
  5.   * @Date 2021/1/18 12:05
  6.   * @Version 1.0
  7.   **/
  9.   package config
  11.   import (
  12.       "github.com/tal-tech/go-queue/dq"
  13.       "github.com/tal-tech/go-zero/core/service"
  15.   )
  17.   type Config struct {
  18.       service.ServiceConf
  19.       DqConf dq.DqConf
  20.   }

Handler/router.go : Responsible for multi-task registration

  1.   /**
  2.   * @Description Register job
  3.   * @Author Mikael
  4.   * @Email 13247629622@163.com
  5.   * @Date 2021/1/18 12:05
  6.   * @Version 1.0
  7.   **/
  8.   package handler
  10.   import (
  11.       "context"
  12.       "github.com/tal-tech/go-zero/core/service"
  13.       "job/internal/logic"
  14.       "job/internal/svc"
  15.   )
  17.   func RegisterJob(serverCtx *svc.ServiceContext,group *service.ServiceGroup)  {
  19.       group.Add(logic.NewProducerLogic(context.Background(),serverCtx))
  20.       group.Add(logic.NewConsumerLogic(context.Background(),serverCtx))
  22.       group.Start()
  24.   }

ProducerLogic: One of the job business logic

  1.   /**
  2.   * @Description Producer task
  3.   * @Author Mikael
  4.   * @Email 13247629622@163.com
  5.   * @Date 2021/1/18 12:05
  6.   * @Version 1.0
  7.   **/
  8.   package logic
  10.   import (
  11.       "context"
  12.       "github.com/tal-tech/go-queue/dq"
  13.       "github.com/tal-tech/go-zero/core/logx"
  14.       "github.com/tal-tech/go-zero/core/threading"
  15.       "job/internal/svc"
  16.       "strconv"
  17.       "time"
  18.   )
  22.   type Producer struct {
  23.       ctx    context.Context
  24.       svcCtx *svc.ServiceContext
  25.       logx.Logger
  26.   }
  28.   func NewProducerLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Producer {
  29.       return &Producer{
  30.           ctx:    ctx,
  31.           svcCtx: svcCtx,
  32.           Logger: logx.WithContext(ctx),
  33.       }
  34.   }
  36.   func (l *Producer)Start()  {
  38.       logx.Infof("start  Producer \n")
  39.       threading.GoSafe(func() {
  40.           producer := dq.NewProducer([]dq.Beanstalk{
  41.               {
  42.                   Endpoint: "localhost:7771",
  43.                   Tube:     "tube1",
  44.               },
  45.               {
  46.                   Endpoint: "localhost:7772",
  47.                   Tube:     "tube2",
  48.               },
  49.           })
  50.           for i := 1000; i < 1005; i++ {
  51.               _, err := producer.Delay([]byte(strconv.Itoa(i)), time.Second * 1)
  52.               if err != nil {
  53.                   logx.Error(err)
  54.               }
  55.           }
  56.       })
  57.   }
  59.   func (l *Producer)Stop()  {
  60.       logx.Infof("stop Producer \n")
  61.   }

Another job business logic

  1.   /**
  2.   * @Description Consumer task
  3.   * @Author Mikael
  4.   * @Email 13247629622@163.com
  5.   * @Date 2021/1/18 12:05
  6.   * @Version 1.0
  7.   **/
  8.   package logic
  10.   import (
  11.       "context"
  12.       "github.com/tal-tech/go-zero/core/logx"
  13.       "github.com/tal-tech/go-zero/core/threading"
  14.       "job/internal/svc"
  15.   )
  17.   type Consumer struct {
  18.       ctx    context.Context
  19.       svcCtx *svc.ServiceContext
  20.       logx.Logger
  21.   }
  23.   func NewConsumerLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Consumer {
  24.       return &Consumer{
  25.           ctx:    ctx,
  26.           svcCtx: svcCtx,
  27.           Logger: logx.WithContext(ctx),
  28.       }
  29.   }
  31.   func (l *Consumer)Start()  {
  32.       logx.Infof("start consumer \n")
  34.       threading.GoSafe(func() {
  35.           l.svcCtx.Consumer.Consume(func(body []byte) {
  36.               logx.Infof("consumer job  %s \n" ,string(body))
  37.           })
  38.       })
  39.   }
  41.   func (l *Consumer)Stop()  {
  42.       logx.Infof("stop consumer \n")
  43.   }

svc/servicecontext.go

  1.   /**
  2.   * @Description Configuration
  3.   * @Author Mikael
  4.   * @Email 13247629622@163.com
  5.   * @Date 2021/1/18 12:05
  6.   * @Version 1.0
  7.   **/
  8.   package svc
  10.   import (
  11.       "job/internal/config"
  12.       "github.com/tal-tech/go-queue/dq"
  13.   )
  15.   type ServiceContext struct {
  16.       Config config.Config
  17.       Consumer      dq.Consumer
  18.   }
  20.   func NewServiceContext(c config.Config) *ServiceContext {
  21.       return &ServiceContext{
  22.           Config: c,
  23.           Consumer: dq.NewConsumer(c.DqConf),
  24.       }
  25.   }

main.go startup file

  1.   /**
  2.   * @Description Startup file
  3.   * @Author Mikael
  4.   * @Email 13247629622@163.com
  5.   * @Date 2021/1/18 12:05
  6.   * @Version 1.0
  7.   **/
  8.   package main
  10.   import (
  11.       "flag"
  12.       "fmt"
  13.       "github.com/tal-tech/go-zero/core/conf"
  14.       "github.com/tal-tech/go-zero/core/logx"
  15.       "github.com/tal-tech/go-zero/core/service"
  16.       "job/internal/config"
  17.       "job/internal/handler"
  18.       "job/internal/svc"
  19.       "os"
  20.       "os/signal"
  21.       "syscall"
  22.       "time"
  23.   )
  26.   var configFile = flag.String("f", "etc/job.yaml", "the config file")
  28.   func main() {
  29.       flag.Parse()
  31.       var c config.Config
  32.       conf.MustLoad(*configFile, &c)
  33.       ctx := svc.NewServiceContext(c)
  35.       group := service.NewServiceGroup()
  36.       handler.RegisterJob(ctx,group)
  38.       ch := make(chan os.Signal)
  39.       signal.Notify(ch, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT)
  40.       for {
  41.           s := <-ch
  42.           logx.Info("get a signal %s", s.String())
  43.           switch s {
  44.           case syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT:
  45.               fmt.Printf("stop group")
  46.               group.Stop()
  47.               logx.Info("job exit")
  48.               time.Sleep(time.Second)
  49.               return
  50.           case syscall.SIGHUP:
  51.           default:
  52.               return
  53.           }
  54.       }
  55.   }