当前位置： > > > > RabbitMQ 消费者性能 – 预取与并发
来源：stackoverflow
2024-04-22 16:09:33
0浏览
收藏
目前上已经有很多关于Golang的文章了，自己在初次阅读这些文章中，也见识到了很多学习思路；那么本文《RabbitMQ 消费者性能 – 预取与并发》，也希望能帮助到大家，如果阅读完后真的对你学习Golang有帮助，欢迎动动手指，评论留言并分享~
问题内容
我有一个 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 来说，为每个工作线程生成一个新通道不是很戏剧性吗？
那么，问题是，哪种变体更好？更好的性能、更好的内存使用等。
那么，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 通道消费操作是 ，所以如果你只做一些消费操作，那么在 goruntine 之间共享通道是安全的。
到这里，我们也就讲完了《RabbitMQ 消费者性能 – 预取与并发》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注公众号，带你了解更多关于的知识点！
