深入浅出RabbitMQ：顺序消费、死信队列和延时队列

本文介绍: RabbitMQ 是一个开源的消息中间件，它实现了高级消息队列协议（AMQP），同时提供了各种重要组件来支持消息的生产、传输和消费。

大家好，我是小❤，一个漂泊江湖多年的 985 非科班程序员，曾混迹于国企、互联网大厂和创业公司的后台开发攻城狮。

上篇文章（应对流量高峰的利器——消息中间件）中，我们已经介绍了消息中间件的用途，主要用作：解耦、削峰、异步通信、应用解耦，并介绍了业界常用的几种消息中间件，优劣对比和使用场景。

在今天的文章中，我们来聊一聊 RabbitMQ，这是小 ❤ 在工作中用的最早的消息中间件，主要用于大量数据的异步消费。

RabbitMQ 是一个开源的消息中间件，它实现了高级消息队列协议（AMQP），同时提供了各种重要组件来支持消息的生产、传输和消费。

说到消息中间件，很多人首先想到的就是 Kafka，但 RabbitMQ 也是许多金融或互联网公司构建可靠、可伸缩和高性能系统的首选。

如上图所示，有三条业务消息分别是删除、增加和修改操作，但是 Consumer 没有按顺序消费，最终存储的顺序是增加、修改和删除，就会发生数据错乱。

在分布式的场景下如果难以保证各个服务器的入队顺序，则可以加分布式锁的方式来解决。或者在业务生产方的消息里带上消息递增 ID，以及消息产生的时间戳。

在 RabbitMQ 的消息会保存在队列（Queue）中，在同一个队列里的消息是先进先出（FIFO）的，这个由 RabbitMQ 来帮我们保证顺序。

在声明队列时，我们可以通过 x-max-priority 属性来设置队列的最大优先级，或通过 Priority 属性来设置消息的优先级，从 1~10。

// 队列属性
props := make(map[string]interface{})
// 设置队列最大优先级
props["x-max-priority"] = 10

ch.Publish(
   "tizi365",     // 交换机
   "", // 路由参数
   false,
   false,
   amqp.Publishing{
       Priority:5, // 设置消息优先级
       DeliveryMode:2,  // 消息投递模式，1代表非持久化，2代表持久化，
       ContentType: "text/plain",
       Body:       []byte(body),
  })

RabbitMQ 里，当消息在队列中变成死信（消费者无法正常处理的消息）之后，它会被重新投递到一个交换机上（即死信交换机），死信交换机上绑定的消费队列就是死信队列。

当死信产生时，如果我们定义了一个死信交换机（其实就是一个普通的交换机，只是用于处理死信，所以叫死信交换机），然后在死信交换机上绑定了一个队列（称作死信队列）。

最后，如果死信队列有消费者监听时，死信消息的处理就会和正常业务消息一样，从交换机到队列，再由死信消费者（监听死信队列的消费者）正常消费。

RabbitMQ 本身不支持延时队列，但是我们可以通过 RabbitMQ 的插件 rabbitmq-delayed-message-exchange，或者使用 死信队列 + 消息过期 的方式来实现。

Github 地址：

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

从 github 的 release 页面的 assets, 下载 rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez 文件，把文件放到 rabbitmq 插件目录（plugins目录）

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

通过 x-delayed-type 设置自定义交换机属性，支持发送延迟消息：

   props := make(map[string]interface{})
   //关键参数，支持发送延迟消息
   props["x-delayed-type"] = "direct"

   // 声明交换机
   err = ch.ExchangeDeclare(
       "delay.queue",   // 交换机名字
       "fanout", // 交换机类型
       true,     // 是否持久化
       false,    
       false,
       false,
       props,      // 设置属性
  )

       msgHeaders := make(map[string]interface{})
       // 通过消息头，设置消息延迟时间，单位毫秒
       msgHeaders["x-delay"] = 6000

       err = ch.Publish(
           "delay.queue",     // 交换机名字
           "", // 路由参数
           false,
           false,
           amqp.Publishing{
               Headers:msgHeaders, // 设置消息头
               ContentType: "text/plain",
               Body:       []byte(body),
          })

然后创建定时过期的消息，比如订单支付的时间为 30min，则将消息的 TTL（最大存活时间）设置为 30min，将消息放到一个没有消费者消费的队列中，当消息过期后就会成为死信。

如果没有支付，就取消订单，修改订单状态为待下单。如果已经支付，就将商品状态修改为已完成，并丢掉这条死信消息。