部署rabbitmq和go-amqp使用

部署rabbitmq

直接使用docker部署一个单节点的mq：

登录dockerhub查看rabbitmq镜像：

下载x-management镜像，此版本带管理后台。

docker pull rabbitmq:3.9.4-management

docker run -d --name rabbitmq3.9.4-management --hostname rabbitmq \
-p 5672:5672 -p 15672:15672 \
-v /var/lib/rabbitmq:/var/lib/rabbitmq \
-e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin123 \
rabbitmq:3.9.4-management

浏览器打开ip:15672访问并登录：

部署完毕。

配置rabbitmq

修改配置

默认配置在一般情况下是够用的。

点击查看github上的配置模版，里边有详细的配置项描述。

分配帐号

Step1：创建test用户

Step2：创建virtual host

Step3：用户授权virtual host

Step4：查看用户权限

使用 Golang Rabbitmq 客户端

这里使用 Go amqp 库来实现最简单的消息模型。

Step1：安装 amqp

1	go get github.com/streadway/amqp

Step2：定义消息发布者 send.go 和消息消费者 receive.go

send.go:

package main

import (
 "log"

 "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
 if err != nil {
  log.Fatalf("%s: %s", msg, err)
 }
}

func main() {
 // 连接 Rabbitmq 服务
 conn, err := amqp.Dial("amqp://test:test123@192.168.1.10:5672//test")
 failOnError(err, "Failed to connect to RabbitMQ")
 defer conn.Close()

 // 创建管道
 ch, err := conn.Channel()
 failOnError(err, "Failed to open a channel")
 defer ch.Close()

 // 声明队列hello并发送消息
 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")

 body := "Hello ashin!"
 err = ch.Publish(
  "",     // exchange
  q.Name, // routing key
  false,  // mandatory
  false,  // immediate
  amqp.Publishing{
   ContentType: "text/plain",
   Body:        []byte(body),
  })
 failOnError(err, "Failed to publish a message")
 log.Printf(" [x] Sent %s", body)
}

receive.go:

package main

import (
 "log"

 "github.com/streadway/amqp"
)

func failOnErr(err error, msg string) {
 if err != nil {
  log.Fatalf("%s: %s", msg, err)
 }
}

func main() {
 // 连接 Rabbitmq 服务
 // 注意上面我们创建的virtual host为 /test，链接中要写成//test
 conn, err := amqp.Dial("amqp://test:test123@192.168.1.10:5672//test")
 failOnErr(err, "Failed to connect to RabbitMQ")
 defer conn.Close()

 // 创建管道
 ch, err := conn.Channel()
 failOnErr(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
 )
 failOnErr(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
 )
 failOnErr(err, "Failed to register a consumer")

 forever := make(chan bool)

 go func() {
  for d := range msgs {
   log.Printf("Received a message: %s", d.Body)
  }
 }()

 // 永久阻塞
 log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
 <-forever
}