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

Question

我有一个 Go 应用程序处理来自单个 RabbitMQ 队列的事件。我使用github.com/streadway/amqpRabbitMQ 客户端库。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这里会并行调用吗？但这是 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之间共享通道是安全的。