# RabbitMQ使用总结(二)

作者: littlejian | 来源:发表于2020-06-29 23:10 被阅读0次

rabbitmq.jpeg

在上一节中我们主要介绍了为什么要使用消息队列、常见的消息队列对比分析以及初步认识了RabbitMQ的整体架构和基本概念。那么在这一节中，主要是从代码的角度出发介绍一下如何使用RabbitMQ以及讲一下在实际项目中踩的坑。

这里我用的是Go语言，连接驱动采用的是开源的库amqp

github地址：https://github.com/streadway/amqp

生产者

建立连接

// 建立连接
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
failOnError(err, "Failed to connect to RabbitMQ")
defer conn.Close()

创建channel

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

创建队列
```
// 创建队列
 q, err := ch.QueueDeclare(
     /*name*/ "queue", // 队列名称
     /*durable*/ false, // 是否持久化
     /*autoDelete*/ false, // 是否自动删除
     /*exclusive*/ false, // 排他
     /*noWait*/ false, // 是否等待服务器确认
     /*args*/ nil, // 其他配置
 )
 failOnError(err, "Failed to declare a queue")
```
参数说明：
- exclusive
  
  排他队列只对首次创建它的连接可见，排他队列是基于连接 (Connection) 可见的，并且该连接内的所有信道 (Channel) 都可以访问这个排他队列，在这个连接断开之后，该队列自动删除，由此可见这个队列可以说是绑到连接上的，对同一服务器的其他连接不可见。
  同一连接中不允许建立同名的排他队列的
  这种排他优先于持久化，即使设置了队列持久化，在连接断开后，该队列也会自动删除。
  非排他队列不依附于连接而存在，同一服务器上的多个连接都可以访问这个队列。
- autoDelete
  
  为 true 则设置队列为自动删除。
  自动删除的前提是：至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。
  不能把这个参数错误地理解为: 当连接到此队列的所有客户端断开时，这个队列自动删除，因为生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列。
- noWait
  
  当 noWait 为 true 时，声明时无需等待服务器的确认。
  该通道可能由于错误而关闭。添加一个 NotifyClose 侦听器应对任何异常。

创建交换机

// 创建交换机
ch.ExchangeDeclare(
   /*name*/ "exchange", // 交换机名称
   /*kind*/ "fanout", // 交换机类型
   /*durable*/ true, // 是否持久化
   /*autoDelete*/ false, // 是否自动删除
   /*internal*/ false, // 是否是内置交换机
   /*noWait*/ false, // 是否等待确认
   /*args*/ nil) // 其他配置
failOnError(err, "Failed to declare a exchange")

参数说明：

internal:

内置交换器是一种特殊的交换器，这种交换器不能直接接收生产者发送的消息，
只能作为类似于队列的方式绑定到另一个交换器，来接收这个交换器中路由的消息，
内置交换器同样可以绑定队列和路由消息，只是其接收消息的来源与普通交换器不同。

绑定交换机和队列

// 绑定交换机和队列
 err = ch.QueueBind(
     /*name*/ q.Name, // 队列名称
     /*key*/ "", // Routing Key 因为采用的是fanout模式，所以这里为空
     /*exchange*/ "exchange", // 交换机名称
     /*noWait*/ false, // 是否等待确认
     /*args*/ nil) // 其他配置
 failOnError(err, "Failed to bind")

发送消息
```
// 发送消息
 body := "Hello World!"
 err = ch.Publish(
     /*exchange*/ "exchange", // 交换机名称
     /*key*/ "", // routing key
     /*mandatory*/ false, // 是否为无法路由的消息进行返回处理
     /*immediate*/ false, // 是否对路由到无消费者队列的消息进行返回处理 RabbitMQ 3.0 废弃
     amqp.Publishing{
         ContentType: "text/plain",
         Body:        []byte(body),
         DeliveryMode: 2,  // 2代表着消息持久化，1代表否
     })
 log.Printf(" [x] Sent %s", body)
 failOnError(err, "Failed to publish a message")
```
参数说明：
- mandatory
  
  消息发布的时候设置消息的 mandatory 属性用于设置消息在发送到交换器之后无法路由到队列的情况对消息的处理方式，
  设置为 true 表示将消息返回到生产者，否则直接丢弃消息。
- immediate
  
  参数告诉服务器至少将该消息路由到一个队列中，否则将消息返回给生产者。imrnediate 参数告诉服务器，如果该消息关联的队列上有消费者，则立刻投递：如果所有匹配的队列上都没有消费者，则直接将消息返还给生产者，不用将消息存入队列而等待消费者了。

如果在客户端提前创建了交换机和队列并且也绑定在一起了，那么可以省略3、4、5步骤，直接发送消息即可。

发送了一条消息，如果这条消息还没有被消费掉，那么在客户端就能看到这条消息：

[图片上传失败...(image-12925a-1593443389768)]

消费者

消费者和生产者步骤类似，这里就贴一下demo：

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() {
   // 建立连接
   conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
   failOnError(err, "Failed to connect to RabbitMQ")
   defer conn.Close()

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

   // 创建队列
   q, err := ch.QueueDeclare(
      "queue", // 队列名称
      true,    // 是否持久化
      false,   // 是否自动删除
      false,   // 排他
      false,   // 是否等待确认
      nil,     // 其他配置
   )
   failOnError(err, "Failed to declare a queue")

   msgs, err := ch.Consume(
      q.Name, // 队列名称
      "",     // consumer
      true,   // 是否自动ack确认
      false,  // 排他
      false,  // no-local 设置为 true 则表示不能将同一个 Connection 中生产者发送的消息传送给这个 Connection 中的消费者
      false,  // 是否等待确认
      nil,    // 其他配置
   )
   failOnError(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
}

go run一下就能接收生产者发送的消息

image-20200627000417465.png

持久化

RabbitMQ 持久化包含3个部分

exchange 持久化，在声明时指定 durable 为 true
queue 持久化，在声明时指定 durable 为 true
message 持久化，在投递时指定 delivery_mode=2（1是非持久化）

这里需要注意以下几点：

queue 的持久化能保证本身的元数据不会因异常而丢失，但是不能保证内部的 message 不会丢失。要确保 message 不丢失，还需要将 message 也持久化。

我之前就是因为没有指定message 持久化，重启了RabbitMQ就发送数据不见了......

如果 exchange 和 queue 都是持久化的，那么它们之间的 binding 也是持久化的。
如果 exchange 和 queue 两者之间有一个持久化，一个非持久化，就不允许建立绑定。
一旦确定了 exchange 和 queue 的 durable，就不能修改了。如果非要修改，唯一的办法就是删除原来的 exchange 或 queue 后，然后重新创建。

数据丢失了怎么办

我们使用消息队列总不能把数据弄丢了吧，这是基本原则。

这里数据丢失主要分三种情况：生产者弄丢了数据、RabbitMQ弄丢了数据以及消费者弄丢了数据。

生产者弄丢了数据

生产者将数据发送到 RabbitMQ 的时候，可能数据就在半路给搞丢了，因为网络问题啥的，都有可能。为了避免这种情况发生，主要有以下两点解决方案：
1. 利用RabbitMQ事物机制
  
  这个呢就是生产者发送数据之前开启 RabbitMQ 事务ch.Tx() ，然后发送消息，如果消息没有成功被 RabbitMQ 接收到，那么生产者会收到异常报错，此时就可以回滚事务 ch.TxRollback() ，然后重试发送消息；如果收到了消息，那么可以提交事务 ch.TxCommit() 。
```
// 发送消息
body := "Hello World!"
ch.Tx()
err = ch.Publish(
   /*exchange*/ "exchange", // 交换机名称
   /*key*/ "", // routing key
   /*mandatory*/ false, // 是否为无法路由的消息进行返回处理
   /*immediate*/ false, // 是否对路由到无消费者队列的消息进行返回处理 RabbitMQ 3.0 废弃
   amqp.Publishing{
      ContentType:  "text/plain",
      Body:         []byte(body),
      DeliveryMode: 2, // 2代表着消息持久化，1代表否
   })
if err != nil {
   log.Printf("tx rollback")
   ch.TxRollback()
} else {
   log.Printf(" [x] Sent %s", body)
   ch.TxCommit()
}
```
2. 采用confirm机制
  
  在生产者那里设置开启 confirm 模式之后，你每次写的消息都会分配一个唯一的 id，然后如果写入了 RabbitMQ 中，RabbitMQ 会给你回传一个 ack 消息，告诉你说这个消息 ok 了。如果 RabbitMQ 没能处理这个消息，ack=false，告诉你这个消息接收失败，你可以重试。而且你可以结合这个机制自己在内存里维护每个消息 id 的状态，如果超过一定时间还没接收到这个消息的回调，那么你可以重发。
```
confirm := ch.NotifyPublish(make(chan amqp.Confirmation))
ch.Confirm(false)
go func() {
   ch.Publish("exchange", "", false, false, amqp.Publishing{Body: []byte("pub 1")})
   ch.Publish("exchange", "", false, false, amqp.Publishing{Body: []byte("pub 2")})
   ch.Publish("exchange", "", false, false, amqp.Publishing{Body: []byte("pub 3")})
   ch.Publish("exchange", "", false, false, amqp.Publishing{Body: []byte("pub 4")})
}()

for i := 0; i < 4; i++ {
   ack := <-confirm
   if ack.Ack {
      // 消息成功发送
   } else {
      // 消息未成功发送
   }
   fmt.Println(ack.DeliveryTag)
}
```
对比这两种方法，我们一般采取第二种方案，因为第一种是同步的，你提交一个事务之后会阻塞在那儿。但是 confirm 机制是异步的，你发送个消息之后就可以发送下一个消息，然后那个消息 RabbitMQ 接收了之后会异步回调你的一个接口通知你这个消息接收到了。
RabbitMQ弄丢了数据

就是 RabbitMQ 自己弄丢了数据，这个你必须开启 RabbitMQ 的持久化，就是消息写入之后会持久化到磁盘，哪怕是 RabbitMQ 自己挂了，恢复之后会自动读取之前存储的数据，一般数据不会丢。除非极其罕见的是，RabbitMQ 还没持久化，自己就挂了，可能导致少量数据丢失，但是这个概率较小。

如果碰到这种情况怎么办呢？

持久化可以跟生产者那边的 confirm 机制配合起来，只有消息被持久化到磁盘之后，才会通知生产者 ack 了，所以哪怕是在持久化到磁盘之前，RabbitMQ 挂了，数据丢了，生产者收不到 ack ，你也是可以自己重发的。
消费者弄丢了数据

RabbitMQ 如果丢失了数据，主要是因为你消费的时候，刚消费到，还没处理，结果进程挂了，比如重启了，那么就尴尬了，RabbitMQ 认为你都消费了，这数据就丢了。

这个时候就需要RabbitMQ提供的ack机制，消费者消费完成后会发送一个ack,RabbitMQ接收到了才会去删除数据，否则保留数据。

前提是关闭RabbitMQ自动ack：
image-20200627125028885.png
```
messages, err := rabbitmq.Consume()
if err != nil {
   logger.ErrorF("[Consume] rabbitmq Consume fail,err: %s", err.Error())
}
go func() {
   ch := make(chan int, 50)
   for msg := range messages {
      go func(m amqp.Delivery) {
         defer func() {
            <-ch
         }()
         ch <- 1
         if err = handleMessage(m.Body); err == nil {
            m.Ack(false) // 发送ack
         } else {
            logger.ErrorF("[Consume] handleMessage fail,err: %s", err.Error())
         }
      }(msg)
   }
}()
```
- m.Ack(true)的时候代表同一channe，此传递和所有先前未确认的传递将一起被确认，这在批量处理的时候会用到。
- m.Ack(false)的时候就是一个一个确认。

总结

本节主要是分享了RabbitMQ在Go里的基本使用以及我们如何去处理消息丢失的一个问题。

当然了还有其他很多更深层次的问题，比如说我们如何去确保消息队列的高可用、如何去确保消息的顺序性、遇到消息积压等等等等。因为呢本人对于RabbitMQ也没有更深入的了解，后续如果发现比较值得学习的地方再来给大家分享。

# RabbitMQ使用总结(二)

生产者

消费者

持久化

数据丢失了怎么办

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Golang 入门资料+笔记