登录
首页 >  Golang >  Go问答

RabbitMQ 消费者性能 - 预取与并发

来源:stackoverflow

时间:2024-04-22 16:09:33 439浏览 收藏

目前golang学习网上已经有很多关于Golang的文章了,自己在初次阅读这些文章中,也见识到了很多学习思路;那么本文《RabbitMQ 消费者性能 - 预取与并发》,也希望能帮助到大家,如果阅读完后真的对你学习Golang有帮助,欢迎动动手指,评论留言并分享~

问题内容

我有一个 go 应用程序处理来自单个 rabbitmq 队列的事件。我使用 github.com/streadway/amqp rabbitmq 客户端库。

go 应用程序在大约 2-3 秒内处理每条消息。如果我从内存中输入消息,则可以并行处理大约 1000 条甚至更多消息。 但不幸的是,rabbitmq 的性能更差。 所以,我想更快地消费队列中的消息。

所以,问题是:如何使用 github.com/streadway/amqp 以最有效的方式消费消息?

据我了解,有两种方法:

  1. 设置高预取

    https://godoc.org/github.com/streadway/amqp#channel.qos。

    使用单个消费者 goroutine

    示例代码:

conn, err := amqp.dial("amqp://guest:guest@localhost:5672/")
failonerror(err, "failed to connect to rabbitmq")
defer conn.close()

ch, err := conn.channel()
failonerror(err, "failed to open a channel")
defer ch.close()

ch.qos(
        10000,           // prefetch count
        0,               // prefetch size
        false,           // global
    )

msgs, err := ch.consume(
  q.name, // queue
  "",     // consumer
  false,  // no auto-ack
  false,  // exclusive
  false,  // no-local
  false,  // no-wait
  nil,    // args
)

for d := range msgs {
  log.printf("received a message: %s", d.body)
  err:= processmessage(d)
  if err != nil {
      log.printf("%s : while consuming task", err)
      d.nack(false, true)
  } else {
      d.ack(false)
  }
  continue // consume other messages
}

但是 processmessage 会在这里并行调用吗?

  1. 产生许多渠道并使用多个消费者
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
failOnError(err, "Failed to connect to RabbitMQ")
defer conn.Close()
var i = 0
for i = 0; i<=100; i++ {
  go func(){
      ch, err := conn.Channel()
      failOnError(err, "Failed to open a channel")
      defer ch.Close()

      ch.Qos(
            10,           // prefetch count
            0,               // prefetch size
            false,           // global
    )

      msgs, err := ch.Consume(
        q.Name, // queue
        "",     // consumer
        false,  // NO auto-ack
        false,  // exclusive
        false,  // no-local
        false,  // no-wait
        nil,    // args
      )

      for d := range msgs {
        log.Printf("Received a message: %s", d.Body)
        err:= processMessage(d)
        if err != nil {
            log.Printf("%s : while consuming task", err)
            d.Nack(false, true)
        } else {
            d.Ack(false)
        }
        continue // consume other messages
      }
  }()
}

但这对 ram 友好吗?对于 rabbitmq 来说,为每个工作线程生成一个新通道不是很戏剧性吗?

那么,问题是,哪种变体更好?更好的性能、更好的内存使用等。

那么,rabbitmq 的最佳用法是什么?

更新:目前,我遇到了一个情况,我的工作线程耗尽了 vps 上的所有 ram,并且被 oom 杀死。我使用了第二种方法。因此,就我而言,更好的是能够让我的工作人员在工作几分钟后不会被 oom 杀死。

更新2:当worker处理消息失败时的nack,以及当worker处理消息时的ack非常重要。所有消息都必须被处理(其客户分析),但有时工作人员无法处理它,因此它必须 nack 消息将其传递给其他工作人员(目前,一些用于处理消息的第 3 方 api 有时只是返回 503 状态代码,在这种情况下的消息应该传递给其他工作人员或重试)。 所以,不幸的是,使用 auto-ack 不是一个选择。


解决方案


我想每个 processMessage() 都在一个新的 goroutine 中运行。

哪种变体更好?

我更喜欢第一个,因为打开/关闭通道有点昂贵(2 + 2 TCP 数据包)。我认为你的 OOM 问题与太多 gorutine 无关,gorutine 很轻,只需要 5KB 左右。所以问题可能是由您的 processMessage() 引起的。

我认为 github.com/streadway/amqp 通道消费操作是 thread/gorutine-safe,所以如果你只做一些消费操作,那么在 goruntine 之间共享通道是安全的。

到这里,我们也就讲完了《RabbitMQ 消费者性能 - 预取与并发》的内容了。个人认为,基础知识的学习和巩固,是为了更好的将其运用到项目中,欢迎关注golang学习网公众号,带你了解更多关于的知识点!

声明:本文转载于:stackoverflow 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>