RabbitMQ性能测试
作者一直在寻找一个极低延时的消息队列，从目前的测试结果来看，只有nats达到了<1ms的水平，本文旨在测试rabbitmq的国latency，撰文记录并与大家分享。
1. 搭建rabbitmq svr
最方便的方式一定是docker，一行命令搞定，这里需要注意的是，没有用rabbitmq:latest，原因是因为这个版本不支持web管理
docker run -d --name rabbitmq -p 5673:5672 -p 15673:15672 docker.io/rabbitmq:3-management
启动后，在浏览器中访问http://localhost:15673/，可以打开管理界面，默认guest:guest
当然，如果希望修改用户名和密码的话，可以用下面的方式启动：
docker run -d -p 15673:15672 -p 5673:5672 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin --name rabbitmq --hostname=rabbitmqhostone docker.io/rabbitmq:3-management
docker exec -it rabbitmq /bin/bash cd etc/rabbitmq/ vim rabbitmq.config {rabbit,[{tcp_listeners,[5672]},{loopback_users,["admin"]}]} rabbitmqctl add_user admin admin rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*" rabbitmqctl set_user_tags admin administrator rabbitmqctl list_users rabbitmqctl list_permissions -p /
2. 安装amqp库
方便起见，本次选用go client。
首先下载amqp的go库 https://github.com/streadway/amqp
https://github.com/streadway/amqp.git
下载完成后，放到GOROOT指定目录下，作者的目录对应/usr/local/go/src/github.com/streadway/amqp
3. 下载客户端库
下载https://github.com/rabbitmq/rabbitmq-tutorials
进入go子目录，go build send.go，go build receive.go，测试可以正常的发送＼接收。
下面在代码中记录发送和接收的毫秒值，用于计算latency。
实测结果，同机测试时延<=1毫秒，可以满足业务的需要。
send.go
package main
import (
"log"
"github.com/streadway/amqp"
"strconv"
"time"
)
func failOnError(err error, msg string) {
if err != nil {
log.Fatalf("%s: %s", msg, err)
}
}
func main() {
conn, err := amqp.Dial("amqp://guest:guest@localhost:5673/")
failOnError(err, "Failed to connect to RabbitMQ")
defer conn.Close()
ch, err := conn.Channel()
failOnError(err, "Failed to open a channel")
defer ch.Close()
q, err := ch.QueueDeclare(
"hello", // name
false, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
failOnError(err, "Failed to declare a queue")
for i:=0; i < 1000; i++ {
time.Sleep(10 * time.Millisecond)
now := time.Now().UnixNano()/1e6
body := strconv.FormatInt(now, 10)
//log.Printf("%d %s\n", now, body)
err = ch.Publish(
"", // exchange
q.Name, // routing key
false, // mandatory
false, // immediate
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(body),
})
//log.Printf(" [x] Sent %s", body)
failOnError(err, "Failed to publish a message")
}
}receive.go
package main
import (
"log"
"time"
"strconv"
"github.com/streadway/amqp"
)
func failOnError(err error, msg string) {
if err != nil {
log.Fatalf("%s: %s", msg, err)
}
}
func main() {
conn, err := amqp.Dial("amqp://guest:guest@localhost:5673/")
failOnError(err, "Failed to connect to RabbitMQ")
defer conn.Close()
ch, err := conn.Channel()
failOnError(err, "Failed to open a channel")
defer ch.Close()
q, err := ch.QueueDeclare(
"hello", // name
false, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
failOnError(err, "Failed to declare a queue")
msgs, err := ch.Consume(
q.Name, // queue
"", // consumer
true, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
failOnError(err, "Failed to register a consumer")
forever := make(chan bool)
go func() {
for d := range msgs {
now := time.Now().UnixNano()/1e6
str := string(d.Body[:])
send_tm, err2 := strconv.ParseInt(str, 10, 64)
failOnError(err2, "Failed to declare a queue")
log.Printf("Received a message: %s send_tm:%d now:%d cost:%d", d.Body, send_tm, now, now-send_tm)
}
}()
log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
<-forever
}参考：
https://www.rabbitmq.com/download.html
https://github.com/alanxz/rabbitmq-c