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

Question

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

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

但这是 RAM 友好的方法吗？对于 RabbitMQ 来说，为每个工作进程生成一个新通道不是很戏剧性吗？

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

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

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

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

Answer 1

我想每次都processMessage()在一个新的 goroutine 中运行。哪个变体更好？我更喜欢第一个，因为打开/关闭通道有点昂贵（2 + 2 TCP 数据包）。我认为你的 OOM 问题与太多 gorutine 无关，gorutine 很轻，只需要 5KB 左右。所以问题很可能是由你的processMessage().我认为github.com/streadway/amqp通道消费操作是线程/goroutine安全的，因此如果你只做一些消费操作，那么在goruntine之间共享通道是安全的。