Go RabbitMQ （一）

RabbitMQ

简介

RabbitMQ是一个消息代理，用来负责接收和转发消息。

术语

• 生产者：生产者是负责发送消息的
• 队列：队列是RabbitMQ用来存储消息的，受主机内存和磁盘大小的限制，本质上是一个消息的缓冲区。生产者能够将消息发送至队列中，消费者能够从队列中接收到消息
• 消费者：消费者是用来等待接收消息

生产者，消费者，代理能够驻留在不一样主机或同一主机，一个应用能够是生产者也能够是消费者

Hello World

接下来咱们来实现RabbitMQ的“Hello World”，生产者将“Hello World”发送进队列中，消费者将其接收并打印

• RabbitMQ客户端的安装
  • RabbitMQ实现了不少协议，在这里咱们使用的是的AMQP 0-9-1，这是一种用于消息传递的开放通用协议。同时有不少关于RabbitMQ的客户端，在这里咱们使用的是Go amqp客户端
  • 安装: **go get github.com/streadway/amqp

发送

• 链接RabbitMQ

  conn,err := amqp.Dial("amqp://guest:guest@localhost:5672/")
  if err != nil {
      log.Fatal(err)
  }
  defer conn.Close()

  RabbitMQ的链接已经为咱们抽象了socket的链接，同时为咱们处理了协议版本号和身份认证等等

• 建立通道

  ch,err := conn.Channel()
  if err != nil {
      log.Fatal(err)
  }
  defer ch.Close()

  在使用其余API完成任务的时候咱们首先经过以上方式建立通道

• 在开始发送消息以前咱们首先应该声明一个队列。声明队列以后咱们就能够将消息发送至队列当中

  q, err := ch.QueueDeclare(
  "hello", // name
  false,   // durable
  false,   // delete when unused
  false,   // exclusive
  false,   // no-wait
  nil,     // arguments
  )
  if err != nil {
      log.Fatal(err)
  }
  body := "Hello World!"
  err = ch.Publish(
  "",     // exchange
  q.Name, // routing key
  false,  // mandatory
  false,  // immediate
  amqp.Publishing {
  ContentType: "text/plain",
  Body:        []byte(body),
  })
  if err != nil {
      log.Fatal(err)
  }

  队列的声明是一个幂等性操做，若是不存在该队列的话则会建立。此处注意，若是队列存在，修改了队列参数并不会影响已经存在的队列，而且会返回错误。消息内容是一个字节数组，因此咱们必须进行编码

接收

• 链接，建立通道，队列

  在接收端咱们一样须要像发送端同样链接RabbitMQ，建立通道后再建立队列，注意此处队列的建立是跟发送端的队列彻底匹配的。队列在接收端也建立是由于咱们接收端有可能比发送端先启动，因此为了保证咱们要消费的队列存在咱们在此处也进行建立

• 消费消息

  msgs, err := ch.Consume(
  q.Name, // queue
  "",     // consumer
  true,   // auto-ack
  false,  // exclusive
  false,  // no-local
  false,  // no-wait
  nil,    // args
  )
  if err != nil {
      log.Fatal(err)
  }   
  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

  使用通道消费队列中的消息，当队列有消息的时候将会异步的推送给咱们