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

【问题标题】：RabbitMQ consumer perfomance - prefetch vs concurencyRabbitMQ 消费者性能 - 预取与并发
【发布时间】：2019-12-13 23:48:57
【问题描述】：

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

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

那么，问题是：如何使用github.com/streadway/amqp 以最有效的方式消费消息？

据我了解，有两种方法：

设置高预取

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 会在这里并行调用吗？

产生许多渠道并使用多个消费者

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 来说，为每个 worker 生成一个新通道不是很戏剧化吗？

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

那么，这里 RabbitMQ 的最佳用法是什么？

更新：目前，我遇到了一个案例，我的工作人员消耗了 VPS 上的所有 RAM，并被 OOM 杀死。我为此使用了第二种方法。因此，在我的情况下，更好的方法是让我的工人在工作几分钟后不会被 OOM 杀死。

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

【问题讨论】：

定义“更好”。我可以争辩说，编写简单的代码比编写复杂的代码要高效得多，因为它节省了开发人员的时间（这很昂贵）。或者，使用大量内存实际上比让它闲置在未使用的情况下要好。除此之外，我会选择以前的代码变体并在下游使用a fan out pattern。
我更新了问题 - 对我来说更好的是 - 代码使用更少的内存，它可以在同一时间处理更多消息。当我尝试实现它时，扇出模式似乎不起作用，因为 d - amqp 消息，ack 和 nack 似乎不是线程安全的......

标签： go rabbitmq

【解决方案1】：

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

哪个变种更好？

我更喜欢第一个，因为打开/关闭通道有点贵（2 + 2 个 TCP 数据包）。我认为您的OOM问题与gorutine过多无关，gorutine很轻，只需花费大约5KB。所以问题可能是由你的processMessage() 引起的。

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

【讨论】：