《RabbitMQ实战指南》整理（三）RabbitMQ进阶

一、消息何去何从

mandatory和immediate是channe.BasicPublish方法中的两个参数，他们都有当消息不可达时将消息返回给生产者的能力。而备份交换器Alternate Exchange可以将未能被交换器路由的消息存储起来，而不用返回给客户端。

1、mandatory参数

当mandatory参数设为true时，交换器无法根据自身的类型和路由键找到符合条件的队列时，会调用BasicReturn命令将消息返回给生产者，当mandatory参数设置为false时，出现上述情况时，消息将被丢弃。生产者如何获取没有被正确路由到合适队列的消息呢？这里我们可以通过IModel.BasicReturn来实现

{
    ......
    channel.BasicReturn += Channel_BasicReturn;
	channel.BasicPublish("amq.direct", routingKey: "MyRoutKey", mandatory: true, basicProperties: null, body: body);
	......
}   

private static void Channel_BasicReturn(object sender, RabbitMQ.Client.Events.BasicReturnEventArgs e)
{
    ......
}

2、immediate参数

当immediate参数设为true时，如果交换器在将消息路由到队列时发现队列上不存在任何消费者，那么这条消息将不会存入队列中。当与路由键匹配的所有队列都没有消费者时，该消息会通过BasicReturn返回给生产者。即至少将该消息路由到一个队列中，否则将消息返回给生产者。RabbitMQ3.0版本已经去掉了对该参数的支持

3、备份交换器

生产者哎发送消息时如果不设置mandatory参数，那么消息会在未被路由的情况下丢失；如果设置了mandatory参数，那么需要添加BasicReturn的逻辑会变的复杂。使用备份交换器可以将未被路由的消息存储在RabbitMQ中，在需要的时候再去处理这些消息。在声明交换器时，可以通过添加alternate-exchange参数来实现，也可以通过策略Policy的方式实现，两者同时使用前者的优先级更高，会覆盖掉Policy的设置。

在使用时可以声明两个交换器，绑定不同的队列，如消息不能被正确的路由到一个交换器绑定的队列时，就会发送给另外一个交换器，进而发送到其绑定的队列。注意这里的“备份交换器”的交换器类型未fanout，当然使用其他类型也可以，但建议设置为fanout

二、过期时间TTL

1、设置消息的TTL

目前有两种办法可以设置消息的TTL，第一种办法是通过队列的属性进行设置，队列中所有的消息都有相同的过期时间；第二种办法是对消息本身进行单独的设置，每条消息的TTL可以不同。如果两者同时使用，消息的TTL以两者中较小的为准。消息在队列中的生存时间一旦超过设置的TTL，就会变为“死信”。

1、通过队列属性设置消息TTL是在channel.queueDeclare方法中加入x-message-ttl参数实现的，其单位是毫秒

var arg = new ConcurrentDictionary<string, object>();
arg.TryAdd("x-message-ttl", 6000);
channel.QueueDeclare(QueueName, false, false, false, arg);

2、针对每条消息设置TTL的方法是在channel.basicPublish方法中加入expiration的属性参数，单位为毫秒

var properties = channel.CreateBasicProperties();
properties.DeliveryMode = 2;//持久化消息
properties.Expiration = "5000";
channel.BasicPublish(ExchangeName, RoutingKey, properties, Encoding.UTF8.GetBytes(message));

对于第一种设置队列TTL属性的方法，一旦消息过期就会从队列中抹去，而第二种即使消息过期也不会马上从队列中抹去，因为每条消息是否过期是在即将投递到消费者之前判断的。前者所有的过期消息都会在队列的头部，所以只需要定期从头部扫描即可，而后者因为过期时间不同，如果删除所有过期消息需要扫描整个队列，因此在消息投递前判断能有效减少性能的损失

2、设置队列的TTL

通过设置channel.queueDeclare方法中的x-expires参数可以控制队列被自动删除前处于未使用状态的时间，即队列上没有任何的消费者，队列也没有被重新声明，并且在过期时间内也未调用过Basic.Get方法。RabbitMQ会确保在过期时间到达后将队列删除，但不保障有多及时，RabbitMQ重启后持久化队列的过期时间会被重新计算

var arg = new ConcurrentDictionary<string, object>();
arg.TryAdd("x-expires", 6000);
channel.QueueDeclare(QueueName, false, false, false, arg);

三、死信队列

1、DLX全称为Dead-Letter-Exchange，称为死信交换器。当消息在一个队列中变成死信之后，它能被重新发送到另一个交换器中，这个交换器就是DLX，绑定DLX的队列就称为死信队列，消息变成死信一般是由于以下几种情况：①消息被拒绝；②消息过期；③队列到达最大长度

2、DLX和一般的交换器没有区别，它可以在任何队列上被指定，实际上就是设置某个队列的属性。通过在channel.queueDeclare方法中设置x-dead-letter-exchange参数来为这个队列添加DLX。

channel.ExchangeDeclare("dlx_exchange","direct");
var arg = new ConcurrentDictionary<string, object>();
arg.TryAdd("x-dead-letter-exchange", "dlx_exchange");
channel.QueueDeclare(QueueName, false, false, false, arg);

3、也可以为这个DLX指定路由键arg.TryAdd("x-dead-letter-routing-key","dlx-routing-key")，如果没有特殊指定，则使用原队列的路由键

4、对于RabbitMQ来说，DLX是一个非常有用的特性，它可以处理异常情况下，不能被消费者正确消费而被置于死信队列中情况，后续可以分析死信队列中的内容来分析解决异常情况。

四、延迟队列

1、延迟队列存储的对象是对应的延迟消息，即当消息被发送后，并不像让消费者立即拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费，其应用场景可以对应下单后N分钟内未支付，智能电器N分钟后执行工作等情况

2、AMQP协议中并没有直接支持延迟队列的功能，但是通过前面介绍的DLX和TTL可以模拟出延迟队列的功能。生产者可以将消息发送到与交换器绑定的不同队列中，同时配置DLX和相应的死信队列，当消息过期时会转存到相应的死信队列中，消费者则可以订阅这些死信队列（即所谓的延迟队列）进行消费

五、优先级队列

1、顾名思义，具有高优先级的队列具有高的优先权，即具备优先被消费的特权。可以通过x-max-priority参数实现

var arg = new ConcurrentDictionary<string, object>();
arg.TryAdd("x-max-priority", 10);
channel.QueueDeclare(QueueName, false, false, false, arg);

2、同样可以对消息进行优先级的配置，消息的优先级默认最低为0，最大为队列的最大优先级，如下：

var properties = channel.CreateBasicProperties();
properties.Priority = 5;
channel.BasicPublish(ExchangeName, RoutingKey, properties, Encoding.UTF8.GetBytes(message));

六、RPC实现

1、RPC即Remote Procedure Call，即远程过程调用，它是一种通过网络从远程计算机上请求服务，而不需要了解底层网络的技术。RPC的主要功能是让构建分布式计算更容易，在提供强大的远程调用能力时不损失本地调用的语义简洁性。

2、一般在RabbitMQ中进行RPC是很简单的，客户端发送请求消息，服务端回复响应的消息，为了接受响应的消息，我们需要在请求消息中发送一个回调队列

var queueName = channel.QueueDeclare().QueueName;
var props = channel.CreateBasicProperties();
props.ReplyTo = queueName;
channel.BasicPublish("", "rpc_queue", props, Encoding.UTF8.GetBytes(message));
//接受返回的消息并进行处理

3、其中replayTo用来设置一个回调队列；correlationId用来关联请求和调用RPC之后的回复。为每个RPC请求创建一个回调队列是非常低效的，通常会为每个客户端创建一个单一的回调队列，但是接收到回复消息后无法对应是哪一个请求，这时候就用到correlationId这个属性了。此外考虑极端情况，回调队列可能会收到重复消息的情况，客户端需要考虑到这种情况并进行相应的处理，并且RPC请求需要保证其本身是幂等的。

示例-客户端：

using System;
using System.Collections.Concurrent;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Producer
{
    public class Program
    {
        static void Main(string[] args)
        {
            var rpcClient = new RpcClient();

            Console.WriteLine(" [x] Requesting fib(30)");
            var response = rpcClient.Call("30");

            Console.WriteLine(" [.] Got '{0}'", response);
            rpcClient.Close();
        }
    }

    public class RpcClient
    {
        private readonly IConnection _connection;
        private readonly IModel _channel;
        private readonly string _replyQueueName;
        private readonly EventingBasicConsumer _consumer;
        private readonly BlockingCollection<string> _respQueue = new BlockingCollection<string>();
        private readonly IBasicProperties _props;

        public RpcClient()
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };

            _connection = factory.CreateConnection();
            _channel = _connection.CreateModel();
            _replyQueueName = _channel.QueueDeclare().QueueName;
            _consumer = new EventingBasicConsumer(_channel);

            _props = _channel.CreateBasicProperties();
            var correlationId = Guid.NewGuid().ToString();
            _props.CorrelationId = correlationId;
            _props.ReplyTo = _replyQueueName;

            _consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var response = Encoding.UTF8.GetString(body);
                if (ea.BasicProperties.CorrelationId == correlationId)
                {
                    _respQueue.Add(response);
                }
            };
        }

        public string Call(string message)
        {
            var messageBytes = Encoding.UTF8.GetBytes(message);
            _channel.BasicPublish("", "rpc_queue", _props, messageBytes);
            _channel.BasicConsume(consumer: _consumer, queue: _replyQueueName, autoAck: true);
            return _respQueue.Take();
        }

        public void Close()
        {
            _connection.Close();
        }
    }
}

示例-服务端：

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;

namespace Consumer
{
  class Program
  {
      static void Main(string[] args)
      {
          var factory = new ConnectionFactory {HostName = "localhost"};
          using var connection = factory.CreateConnection();
          using var channel = connection.CreateModel();
          channel.QueueDeclare("rpc_queue", false, false, false, null);
          channel.BasicQos(0, 1, false);
          var consumer = new EventingBasicConsumer(channel);
          channel.BasicConsume("rpc_queue", false, consumer);
          Console.WriteLine(" [x] Awaiting RPC requests");

          consumer.Received += (model, ea) =>
          {
              string response = null;

              var body = ea.Body.ToArray();
              var props = ea.BasicProperties;
              var replyProps = channel.CreateBasicProperties();
              replyProps.CorrelationId = props.CorrelationId;

              try
              {
                  var message = Encoding.UTF8.GetString(body);
                  int n = int.Parse(message);
                  Console.WriteLine(" [.] fib({0})", message);
                  response = Fib(n).ToString();
              }
              catch (Exception e)
              {
                  Console.WriteLine(" [.] " + e.Message);
                  response = "";
              }
              finally
              {
                  var responseBytes = Encoding.UTF8.GetBytes(response);
                  channel.BasicPublish("", props.ReplyTo, replyProps, responseBytes);
                  channel.BasicAck(ea.DeliveryTag, false);
              }
          };

          Console.WriteLine(" Press [enter] to exit.");
          Console.ReadLine();
      }

      private static int Fib(int n)
      {
          if (n == 0 || n == 1)
          {
              return n;
          }

          return Fib(n - 1) + Fib(n - 2);
      }
  }
}

七、持久化

1、RabbitMQ的持久化分为三个部分：

• 交换器的持久化：通过在声明交换器时将durable参数设置为true实现，如未设置交换器元数据会丢失，消息不会丢失，但是不能将消息发送到这个交换器中了，一般建议设置为持久化
• 队列的持久化：通过在声明队列时将durable参数设置为true实现，如未设置队列的元数据会丢失，消息也会丢失。队列的持久化可以保证本身的元数据不会因为异常丢失，但并不能保证内部的消息不丢失，除非将消息的投递模式中的deliveryMode属性设置为2
• 消息的持久化：可以通过将消息的投递模式中的deliveryMode属性设置为2进行持久化

2、一般来说队列的持久化和消息的持久化需要配合使用，否则是没有意义的。但是将所有的消息全部设置为持久化会严重影响RabbitMQ的性能，对于可靠性要求不高的消息可以在可靠性和吞吐量之间做权衡

3、将交换器、队列和消息设置为持久化之后并不能保证百分百数据不丢失，如消费者收到消息后未处理就宕机同样会造成数据的丢失，所以一般情况下将autoAck参数设置为false，并进行手动确认。另外RabbitMQ并不是为每条数据进行同步存盘，而是先存在系统缓存中，再调用内核的fsync方法进行批量的存储，如果在这期间发生宕机同样会造成消息的丢失。RabbitMQ可以通过镜像队列机制为其配置副本，主节点挂掉后从节点可以自动切换顶上，虽然仍不能保证数据百分比不丢失，但已经相对靠谱的多

八、生产者确认

当消息的生产者将消息发送出去后，再不进行特殊配置的情况下，是无法知道消息是否到达服务器的，如果发生丢失，即使设置了持久化也无法保证数据的到达，针对这种情况，RabbitMQ提供了两种解决方式：①事务机制；②发送方确认机制

1、事务机制

RabbitMQ中与事务机制相关的方法有三个：TxSelect、TxCommit和TxRollback。TxSelect用于将当前信道设置成事务模式，TxCommit用于提交事务，TxRollback用于事务回滚

开启事务与不开启事务相比多了四个步骤：①客户端发送Tx.Select将信道设置成事务模式；②Broker回复Tx.Select为OK确认已将信道设置成事务模式；③发送完消息后，客户端发送Tx.Commit提交事务；④Broker回复Tx.Commit为OK确认提交事务

2、发送方确认机制

采用事务机制会严重降低RabbitMQ的消息吞吐量，使用发送方确认机制可以轻量级的实现生产者确认的问题。生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都会被指派一个唯一ID，一旦消息被投递到匹配的队列之后，RabbitMQ就会发送一个确认给生产者（包含唯一ID），使得生产者知晓消息已经正确到达目的地了。如果消息和队列是持久化的，确认消息会在写入磁盘后发出。确认消息中的deliveryTag包含了确认消息的序号，而basicAck方法中的multiple参数表示到这个序号之前的所有消息都已得到处理

事务机制会在一条消息发送之后使发送端阻塞，等待RabbitMQ回应后才能继续发送下一条消息；而发送方确认机制是异步的，生产者可以同各国回调方法来处理确认消息，即使消息丢失也可以在回调方法中处理nack命令。如果信道没有开启confirm模式，则调用任何WaitForConfirms方法都会报错，对于没有参数的WaitForConfirms，其返回条件是客户端收到 了相应的ack/nack或者被中断

channel.ConfirmSelect();
channel.BasicPublish("ExchangeName","RoutingKey",null, Encoding.UTF8.GetBytes("xxx"));
if (_channel.WaitForConfirms())
{
   Console.WriteLine("发送失败...");
   //dosomething
}

注：事务机制和confirm机制两者是互斥的，不能共存。Confirm的优势在于不一定需要同步确认，可以使用批量Confirm方法；或者使用回调方法实现异步confirm，如可以使用 channel.BasicNacks和channel.BasicAcks来实现，其中参数DeliveryTag即为confirm模式下用来标记消息的唯一序号

九、消费端要点介绍

消费端可以通过拉模式或者推模式的方法获取并消费消息，消费者处理完消息后需要手动确认消息已被接受，RabbitMQ才能把消息从队列中标记清除，如果因为某些原因无法处理接受到的信息，可以使用channel.BasicNacks或者channel.BasicReject来拒绝掉。对于消费端来说有几点需要注意：①消息分发；②消息顺序性；③弃用QueueingConsumer

1、消息分发

RabbitMQ队列收到消息后会以轮询的方式分发给消费者，但如果不同消费者处理消息的能力差异较大，就会造成部分消费者一直处于忙碌状态，另一部分消费者处于空闲状态，这个时候我们可以使用channel.basicQos方法来限制信道上消费者所能保持的最大未确认消息的数量。需要注意的是Basic.Qos对拉模式的消费方式是无效的。此外该函数的global参数未false时表示信道上新的消费者需要遵循当前设定的prefetchCount，为true表示信道上所有的消费者都要遵从当前设定的prefetchCount

如果在订阅消息之前，即设置了global为true，又设置了为false那么两者都会生效，即每个消费者只能接收到设定为false的Qos的prefetchCount，而两个消费者收到未确认消息的总量不能超过设定为true的Qos的prefetchCount。一般情况下是不建议这么设定的，如无特殊需要，一般设定为false即可

2、消息顺序性

消息的顺序性是指消费者消费到的消息和发送者发布的消息顺序是一致的。在有多个生产者同时发送消息的情况下是无法确定到达Broker的前后顺序的，此外在补偿发送和设定了延迟队列或优先级的情况下，消息的顺序性同样是无法保证的。如果要确保消息的顺序性，需要使用进一步的处理，比如添加全局有序标识等

十、消息传输保障

1、一般消息中间件的消息传输保障分为三个层级：

• 最多一次：消息可能会丢失，但绝不会重复传输；
• 最少一次：消息绝不会丢失，但可能会重复传输；
• 恰好一次：每条消息肯定会被传输一次且仅传输一次；

2、RabbitMQ支持最多一次和最少一次，并且最少一次的投递实现需要考虑以下几个方面：

• 消息生产者需要开启事务机制或confirm确认机制，确保消息的可靠性传输；
• 消息生产者需要配合使用mandatory参数或者备份交换器来确保消息能够从交换器路由到队列中，不会被丢弃；
• 消息和队列都需要进行持久化处理，确保服务器在遇到异常情况时不会造成消息的丢失；
• 消费者在消费消息时需要将autoAck设置为false，并手动确认避免在消费端引起不必要的消息丢失

3、恰好一次的情况是无法保障的，且大多数消息中间件都没有去重机制，一般情况下业务方可以根据业务的特性进行去重，或者是确保自身的幂等性，或者结束Redis等产品做去重处理