网页开发与设计的内容,广州seo网站排名,app门户网站,网站开发工程师需要会写什么前言上篇讨论过消息投递和消息消费过程中如何确保可靠传输#xff0c;也提及到消息到达RabbitMQ中到被消费前也需要可靠的留存#xff0c;可因许多的不确定因素会影响着消息的存在与否。消息中转点生产者发送消息到RabbitMQ中#xff0c;如果交换机根据自身类型和RoutingKey…前言上篇讨论过消息投递和消息消费过程中如何确保可靠传输也提及到消息到达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 rabbitmqdemotest,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 EventHandlerRabbitMQ.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 rabbitmqdemotest,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 Dictionarystring, 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,望技术有成后能回来看见自己的脚步