DotNet Core&RabbitMQ消息存储可靠机制
目录
前言
上篇讨论过消息投递和消息消费过程中如何确保可靠传输，也提及到消息到达RabbitMQ中到被消费前也需要可靠的留存，可因许多的不确定因素会影响着消息的存在与否。
消息中转点
生产者发送消息到RabbitMQ中，如果交换机根据自身类型和RoutingKey能够匹配到队列，则存入相关队列，但当匹配不到队列时，遇到两种情况而使得消息走向不同的方向，消息可能会丢失或是发回给生产者，这取决于生产者对消息的配置。
生产者设置了Mandatory且为true，则消息回退给生产者。
当生产者为设置Mandatory或是设置为false时，为了避免消息丢失，可以由交换机路由给备份交换机负责去搞定存储。
Mandatory
生产者发送消息时，可以设置一个参数mandatory，来决定消息到达RabbitMQ后，如果出现交换机根据自身类型及RoutingKey找不到合适的队列情况下，消息的一个走向。
当mandatory为true时，消息则返回给生产者。
当mandatory为false时，消息则被丢弃。
生产者代码
当在BasicPublish方法参数中设置mandatory为true且队列暂不声明时，仅有一个交换机，消息将会被返回。
var connFactory = new ConnectionFactory
{
HostName = "xxx.xxx.xxx.xxx",
Port = 5672,
UserName = "rabbitmqdemo",
Password = "rabbitmqdemo@test",
VirtualHost = "rabbitmqdemo"
};
using (var conn = connFactory.CreateConnection())
{
using (var channel = conn.CreateModel())
{
var exchangeName = "mandatory_publishsubscribe_exchange";
channel.ExchangeDeclare(exchange: exchangeName, type: "fanout");
while (true)
{
Console.WriteLine("消息内容(exit退出):");
var message = Console.ReadLine();
if (message.Trim().ToLower() == "exit")
{
break;
}
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: exchangeName, routingKey: "", mandatory: true, basicProperties: null, body: body);
Console.WriteLine("消息内容发送完毕:" + message);
}
}
}
生产者发送消息，交换机收到消息但无对应队列，消息被返回。
为了直观的知道消息返回到了生产者，我们可以增加一个监听器，来监听返回的消息。
监听回退消息
当mandatory设置为true，消息回退时可以监听消息
channel.BasicReturn += new EventHandler<RabbitMQ.Client.Events.BasicReturnEventArgs>((sender, e) =>
{
var message = Encoding.UTF8.GetString(e.Body.ToArray());
Console.WriteLine($"收到回退消息：{message}");
});
生产者发送消息，因无匹配队列，消息被返回，可以直观的看到返回的消息。
备份交换机
当mandatory设置为false时，消息被丢失了，这种情况可不太好。可以使用备份交换机来存储原要被丢弃的消息，当需要这些消息的时候，还能拿到这些消息。实际上备份交换机没有什么特殊，和主交换机是一样的只是充当备份的角色。
生产者代码
在创建主交换机的时候，给定参数argument，设置该主交换机的备份交换机，指定备份交换机名称。
然后声明备份交换机并绑定一个队列，用于存储被丢弃的消息。
发送消息时mandatory参数设置为false。
var connFactory = new ConnectionFactory
{
HostName = "xxx.xxx.xxx.xxx",
Port = 5672,
UserName = "rabbitmqdemo",
Password = "rabbitmqdemo@test",
VirtualHost = "rabbitmqdemo"
};
using (var conn = connFactory.CreateConnection())
{
using (var channel = conn.CreateModel())
{
var exchangeName = "aedemo_publishsubscribe_exchange";
var alternateExchangeName = "aedemo_ae_publishsubscribe_exchange";
var arguments = new Dictionary<string, object>
{
{ "alternate-exchange", alternateExchangeName }
};
channel.ExchangeDeclare(exchange: exchangeName, type: "fanout", arguments: arguments);
channel.ExchangeDeclare(exchange: alternateExchangeName, type: "fanout");
var alternateExchangeQueueName = alternateExchangeName + "_worker";
channel.QueueDeclare(queue: alternateExchangeQueueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
channel.QueueBind(queue: alternateExchangeQueueName, exchange: alternateExchangeName, routingKey: "");
while (true)
{
Console.WriteLine("消息内容(exit退出):");
var message = Console.ReadLine();
if (message.Trim().ToLower() == "exit")
{
break;
}
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: exchangeName, routingKey: "", mandatory: false, basicProperties: null, body: body);
Console.WriteLine("消息内容发送完毕:" + message);
}
}
}
启动程序，可以从Web面板中看到主交换机和备份交换机都创建完毕，并且主交换机打上了有AE的标记。
生产者发送消息，经主交换机匹配但无合适队列后，转发给备份交换机，路由到其队列存储。
注：推荐使用Fanout类型的交换机，如果其他比如Direct，当主交换机转发到备份交换机，在进行匹配时候，如果消息给定的RoutingKey没有匹配到相应的队列，消息则会被丢失，这样一来，最初的预想就出现偏差了。
持久化
当RabbitMQ在异常情况下，比如系统宕机、重启、关闭等，可能会导致数据丢失，可靠性降低。针对这种情况，RabbitMQ提供了持久化机制，将消息本身和元数据（队列、交换机、绑定信息）都保存到磁盘中。具体分为三类持久化
交换机持久化
队列持久化
消息持久化
交换机持久化
当RabbitMQ遇到异常情况(如服务重启)后，如果没有设置交换机持久化，那么交换机相关数据则会被丢失，生产者再发送消息到指定交换机时就失败了。
服务重启异常
1、在Web中新建一个交换机，指定非持久化模式。
2、新建一个队列，指定非持久化模式。
3、设置交换机和队列的绑定关系。
4、生产者前部分正常发送消息，中间经服务重启后，交换机、队列及绑定关系都被清除，生产继续发送消息，出现异常。
持久化设置
在声明交换机时可以指定durable参数设置为true(Web面板中也可设置)。
channel.ExchangeDeclare(exchange: exchangeName, type: "fanout", durable: true, arguments: null);
RabbitMQ服务重启，生产者继续发送消息给交换机。
队列持久化
队列的持久化是队列声明时设置durable参数为true，如果队列不持久化，异常情况(如服务重启)后，队列元数据丢失，存储在内的消息也就丢失了。
服务重启异常
1、Web中创建一个交换机并设置为持久化模式。
2、创建一个队列并设置为非持久化模式
3、设置交换机和队列的绑定关系。
4、生产者前部分正常发送消息，中间经服务重启后，队列及绑定关系被清除，生产继续发送消息，匹配队列失败，消息被回退给生产者。
持久化设置
在声明队列时可以指定durable参数设置为true(Web面板中也可设置)。
channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
RabbitMQ服务重启，生产者继续发送消息给交换机。
消息持久化
队列的持久化仅能保证其自身的数据不丢失，而其存储的消息却不能保证不会丢失。
持久化设置
需要对消息消息设置持久化，以确保消息本身不会因异常情况(如服务重启)而丢失。在发送消息时，可以设置消息的基础属性，来支持消息的持久化。
var basicProperties = channel.CreateBasicProperties();
basicProperties.DeliveryMode = 2;// 1非持久化 2持久化
channel.BasicPublish(exchange: exchangeName, routingKey: "", mandatory: true, basicProperties: basicProperties, body: body);
如此一来，当异常情况(如服务重启后)，消息还是存在的。
注：消息持久化会影响性能，仅确保有价值的消息持久化，来权衡可靠与吞吐量。
2022-08-25,望技术有成后能回来看见自己的脚步