延时队列

概述

延时队列是一种特殊的队列，它的元素只有在指定的延时时间到达之后才能被消费。延时队列的实现通常依赖于定时任务，定时任务会定期扫描队列中的元素，将到期的元素移动到另一个队列中，消费者从这个队列中消费元素。

在 go-zero 中我们使用 go-queue 来实现延时队列。

任务目标

  • 了解 go-queue 的基本使用
  • 了解如何在 go-zero 中使用延时队列

准备条件

生产者

dq 生产者很简单，只需要 beanstalkd 地址，tube 即可创建一个 Producer 对象。

  1. type Beanstalk struct {
  2.   Endpoint string // beanstalkd 地址
  3.   Tube     string // tube 名称
  4. }
  6. NewProducer(beanstalks []Beanstalk) Producer

代码示例

  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.     // 延迟 5s 后处理
  24.     _, err := producer.Delay([]byte("hello"), time.Second*5)
  25.     if err != nil {
  26.         fmt.Println(err)
  27.     }
  29.     // 在指定时间点处理
  30.     _, err = producer.At([]byte("hello"), time.Now().Add(time.Second*10))
  31.     if err != nil {
  32.         fmt.Println(err)
  33.     }
  34. }

消费者

配置介绍

  1. type DqConf struct {
  2.   Beanstalks []Beanstalk
  3.   Redis      redis.RedisConf
  4. }
  6. type Beanstalk struct {
  7.   Endpoint string // beanstalkd 地址
  8.   Tube     string // tube 名称
  9. }
延时队列 - 图1参数延时队列 - 图2类型延时队列 - 图3是否必填延时队列 - 图4默认值延时队列 - 图5说明
Beanstalks[]Beanstalk是beanstalkd 配置
RedisRedisConf是redis 配置，详情参考Redis 配置

Beanstalk

延时队列 - 图6参数延时队列 - 图7类型延时队列 - 图8是否必填延时队列 - 图9默认值延时队列 - 图10说明
Endpointstring是beanstalkd 地址
Tubestring是tube 名称

示例

  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. }

参考文献

  1. 《go-beanstalk》
  2. 《beanstalkd》