Delay Queue

Overview

Delay queue is a special queue whose elements can only be consumed after the specified delay has arrived.Delayed queue implementation usually relies on a scheduled task, which regularly scan elements in the queue, move expired elements to another queue, from which consumers consume.

In go-zero we use go-queue to achieve delay queue

Task Targets

  • Learn about the basic usage of go-queue
  • Learn how to use the delay queue in go-zero

Preparing

Producer

dq Producer is simple and needs only beanstalkd addresses, tube can create a producer object.

  1. type Beanstalk struct {
  2.   Endpoint string // beanstalkd server address
  3.   Tube     string // tube name
  4. }
  6. NewProducer(beanstalks []Beanstalk) Producer

Code Example

  1. package main
  3. import (
  4.     "fmt"
  5.     "strconv"
  6.     "time"
  8.     "github.com/zeromicro/go-queue/dq"
  9. )
  11. func main() {
  12.     producer := dq.NewProducer([]dq.Beanstalk{
  13.         {
  14.             Endpoint: "localhost:11300",
  15.             Tube:     "tube",
  16.         },
  17.         {
  18.             Endpoint: "localhost:11301",
  19.             Tube:     "tube",
  20.         },
  21.     })
  23.     _, err := producer.Delay([]byte("hello"), time.Second*5)
  24.     if err != nil {
  25.         fmt.Println(err)
  26.     }
  28.     _, err = producer.At([]byte("hello"), time.Now().Add(time.Second*10))
  29.     if err != nil {
  30.         fmt.Println(err)
  31.     }
  32. }

Consumer

Configure Introduction

  1. type DqConf struct {
  2.   Beanstalks []Beanstalk
  3.   Redis      redis.RedisConf
  4. }
  6. type Beanstalk struct {
  7.   Endpoint string // beanstalkd server address
  8.   Tube     string // tube name
  9. }
Delay Queue - 图1ParamsDelay Queue - 图2DataTypeDelay Queue - 图3Required?Delay Queue - 图4Default valueDelay Queue - 图5Note
Beanstalks[]BeanstalkYESbeanstalkd configuration
RedisRedisConfYESredis configuration, referenceRedis configuration

Beanstalk

Delay Queue - 图6ParamsDelay Queue - 图7DataTypeDelay Queue - 图8Required?Delay Queue - 图9Default valueDelay Queue - 图10Note
EndpointstringYESbeanstalkd address
TubestringYEStube name

Sample

  1. package main
  3. import (
  4.     "fmt"
  6.     "github.com/zeromicro/go-queue/dq"
  7.     "github.com/zeromicro/go-zero/core/stores/redis"
  8. )
  10. func main() {
  11.     consumer := dq.NewConsumer(dq.DqConf{
  12.         Beanstalks: []dq.Beanstalk{
  13.             {
  14.                 Endpoint: "localhost:11300",
  15.                 Tube:     "tube",
  16.             },
  17.             {
  18.                 Endpoint: "localhost:11301",
  19.                 Tube:     "tube",
  20.             },
  21.         },
  22.         Redis: redis.RedisConf{
  23.             Host: "localhost:6379",
  24.             Type: redis.NodeType,
  25.         },
  26.     })
  27.     consumer.Consume(func(body []byte) {
  28.         fmt.Println(string(body))
  29.     })
  30. }

References

  1. go-beanstalk:
  2. Beanstalkd