在开始自研 go-queue 之前,针对以下我们调研目前的开源队列方案:

beanstalkd

beanstalkd 有一些特殊好用功能:支持任务 priority、延时 (delay)、超时重发 (time-to-run) 和预留 (buried),能够很好的支持分布式的后台任务和定时任务处理。如下是 beanstalkd 基本部分:

很幸运的是官方提供了 go client:https://github.com/beanstalkd/go-beanstalk。

但是这对不熟悉 beanstalkd 操作的 go 开发者而言,需要学习成本。

kafka

类似基于 kafka 消息队列作为存储的方案,存储单元是消息,如果要实现延时执行,可以想到的方案是以延时执行的时间作为 topic,这样在大型的消息系统中,充斥大量一次性的 topicdq_1616324404788, dq_1616324417622),当时间分散,会容易造成磁盘随机写的情况。

而且在 go 生态中,

同时考虑以下因素:

所以我们自己基于以上两个基础组件开发了 go-queue

整体设计如下:

【GoCN酷Go推荐】分布式任务 + 消息队列框架 go-queue-LMLPHP

首先在消费场景来说,一个是针对任务队列,一个是消息队列。而两者最大的区别:

所以在背后的基础设施选型上,也是基于这种消费场景。

而从其中 dq的 API 中也可以看出:

// 延迟任务执行
- dq.Delay(msg, delayTime);

// 定时任务执行
- dq.At(msg, atTime);

而在我们内部:

分别介绍dqkq的使用方式:

dq

// [Producer]
producer := dq.NewProducer([]dq.Beanstalk{
    {
        Endpoint: "localhost:11300",
        Tube:     "tube",
    },
    {
        Endpoint: "localhost:11301",
        Tube:     "tube",
    },
})  

for i := 1000; i < 1005; i++ {
    _, err := producer.Delay([]byte(strconv.Itoa(i)), time.Second*5)
    if err != nil {
        fmt.Println(err)
    }
}
// [Consumer]
consumer := dq.NewConsumer(dq.DqConf{
  Beanstalks: []dq.Beanstalk{
    {
      Endpoint: "localhost:11300",
      Tube:     "tube",
    },
    {
      Endpoint: "localhost:11301",
      Tube:     "tube",
    },
  },
  Redis: redis.RedisConf{
    Host: "localhost:6379",
    Type: redis.NodeType,
  },
})
consumer.Consume(func(body []byte) {
  // your consume logic
  fmt.Println(string(body))
})

和普通的 生产者 - 消费者 模型类似,开发者也只需要关注以下:

kq

producer.go:

// message structure
type message struct {
    Key     string `json:"key"`
    Value   string `json:"value"`
    Payload string `json:"message"`
}

pusher := kq.NewPusher([]string{
    "127.0.0.1:19092",
    "127.0.0.1:19093",
    "127.0.0.1:19094",
}, "kq")

ticker := time.NewTicker(time.Millisecond)
for round := 0; round < 3; round++ {
    select {
    case <-ticker.C:
        count := rand.Intn(100)
    // 准备消息
        m := message{
            Key:     strconv.FormatInt(time.Now().UnixNano(), 10),
            Value:   fmt.Sprintf("%d,%d", round, count),
            Payload: fmt.Sprintf("%d,%d", round, count),
        }
        body, err := json.Marshal(m)
        if err != nil {
            log.Fatal(err)
        }

        fmt.Println(string(body))
    // push to kafka broker
        if err := pusher.Push(string(body)); err != nil {
            log.Fatal(err)
        }
    }
}

config.yaml

Name: kq
Brokers:
  - 127.0.0.1:19092
  - 127.0.0.1:19092
  - 127.0.0.1:19092
Group: adhoc
Topic: kq
Offset: first
Consumers: 1

consumer.go:

var c kq.KqConf
conf.MustLoad("config.yaml", &c)

// WithHandle: 具体的处理msg的logic
// 这也是开发者需要根据自己的业务定制化
q := kq.MustNewQueue(c, kq.WithHandle(func(k, v string) error {
  fmt.Printf("=> %s\n", v)
  return nil
}))
defer q.Stop()
q.Start()

dq 不同的是:开发者不需要关注任务类型(在这里也没有任务的概念,传递的都是 message data)。

其他操作和 dq 类似,只是将 业务处理函数 当成配置直接传入消费者中。

在我们目前的场景中,kq 大量使用在我们的异步消息服务;而延时任务,我们除了 dq,还可以使用内存版的 TimingWheel「go-zero 生态组件」。

关于 go-queue 更多的设计和实现文章,可以持续关注我们。欢迎大家去关注和使用。

https://github.com/tal-tech/go-queue

https://github.com/tal-tech/go-zero

欢迎使用 go-zero 并 star 支持我们!

还想了解更多吗?

更多请查看:https://github.com/tidwall/gjson

欢迎加入我们GOLANG中国社区:https://gocn.vip/


《酷Go推荐》招募:


各位Gopher同学,最近我们社区打算推出一个类似GoCN每日新闻的新栏目《酷Go推荐》,主要是每周推荐一个库或者好的项目,然后写一点这个库使用方法或者优点之类的,这样可以真正的帮助到大家能够学习到新的库,并且知道怎么用。


大概规则和每日新闻类似,如果报名人多的话每个人一个月轮到一次,欢迎大家报名!


点击 阅读原文 即刻报名




本文分享自微信公众号 - GoCN(golangchina)。
如有侵权,请联系 [email protected] 删除。
本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

04-02 16:24