Queue

    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

    etc/job.yaml : Configuration file

    Internal/config/config.go: Parse dq corresponding 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.   }

    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"
    15.       "strconv"
    16.       "time"
    17.   )
    21.   type Producer struct {
    22.       ctx    context.Context
    23.       svcCtx *svc.ServiceContext
    24.       logx.Logger
    25.   }
    27.   func NewProducerLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Producer {
    28.       return &Producer{
    29.           ctx:    ctx,
    30.           svcCtx: svcCtx,
    31.           Logger: logx.WithContext(ctx),
    32.       }
    33.   }
    35.   func (l *Producer)Start()  {
    37.       logx.Infof("start  Producer \n")
    38.       threading.GoSafe(func() {
    39.           producer := dq.NewProducer([]dq.Beanstalk{
    40.               {
    41.                   Endpoint: "localhost:7771",
    42.                   Tube:     "tube1",
    43.               },
    44.               {
    45.                   Endpoint: "localhost:7772",
    46.                   Tube:     "tube2",
    47.               },
    48.           })
    49.               _, err := producer.Delay([]byte(strconv.Itoa(i)), time.Second * 1)
    50.               if err != nil {
    51.                   logx.Error(err)
    52.               }
    53.           }
    54.       })
    55.   }
    57.   func (l *Producer)Stop()  {
    58.       logx.Infof("stop Producer \n")
    59.   }

    Another job business logic

    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),

    main.go startup file