RabbitMQ 入门系列:1、MQ的应用场景的选择与RabbitMQ安装。
RabbitMQ 入门系列:2、基础含义:链接、通道、队列、交换机。
RabbitMQ 入门系列:3、基础含义:持久化、排它性、自动删除、强制性、路由键。
RabbitMQ 入门系列:4、基础编码:官方SDK使用:链接创建、单例改造、发送消息、接收消息。
RabbitMQ 入门系列:5、基础编码:交换机的进阶介绍及编码方式。
RabbitMQ 入门系列:6、保障消息:不丢失:发送方、Rabbit存储端、接收方。
RabbitMQ 入门系列:7、保障消息:不重复消费:产生消息的唯一ID。
RabbitMQ 入门系列:8、扩展内容:接收信息时:可否根据RoutingKey过滤监听信息,答案是不能。
RabbitMQ 入门系列:9、扩展内容:死信队列:真不适合当延时队列。
RabbitMQ 入门系列:10、扩展内容:延时队列:延时队列插件及其有限的适用场景。
本篇简单介绍如何保障消息不丢失的处理方式。
主要是通过消息确认或事务,来保障这个过程,下面见具体代码:
using RabbitMQ.Client;using System.Text;using (var channel = Rabbit.Instance.DefaultConnection.CreateModel()){ channel.ConfirmSelect();//开启确认 channel.QueueDeclare("FirstQueue", false, false, false); channel.BasicPublish("", "FirstQueue", false, null, Encoding.UTF8.GetBytes("这是要发送的内容")); if (channel.WaitForConfirms(TimeSpan.FromSeconds(10)))//设置最长超时时间 { //发送确认成功 } else { //超时或失败,需要处理是否重发消息。 }}
using RabbitMQ.Client;using System.Text;using (var channel = Rabbit.Instance.DefaultConnection.CreateModel()){ channel.TxSelect(); channel.QueueDeclare("FirstQueue", false, false, false); channel.BasicPublish("", "FirstQueue", false, null, Encoding.UTF8.GetBytes("这是要发送的内容")); try { channel.TxCommit(); } catch (Exception) { channel.TxRollback(); //处理事务提交失败的逻辑。 }}
对于RabbitMQ端的消息保障,我们人为可以处理的是,设置创建的队列或消息是否持久化。
通过创建持久化的队列或消息,可以保障消息写入硬盘,重启时仍能还原信息。
//第二个参数:是否持久化channel.QueueDeclare("FirstQueue", true, false, false);
接收方主要是通过消息确认,来指示是否收到信息。
var channel = Rabbit.Instance.DefaultConnection.CreateModel();var consumer = new EventingBasicConsumer(channel);consumer.Received += (model, ea) =>{ var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine("收到默认消息 {0}", message);};channel.BasicConsume(queue: "FirstQueue", autoAck: true, consumer: consumer);
在以上代码中,通过指定authAck可以自动回应收到信息。
当然,有需要也可以手动回应:
var channel = Rabbit.Instance.DefaultConnection.CreateModel();var consumer = new EventingBasicConsumer(channel);consumer.Received += (model, ea) =>{ var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine("收到默认消息 {0}", message); try { channel.BasicAck(ea.DeliveryTag, false); } catch (Exception err) { //处理确认失败的情况。 }};channel.BasicConsume(queue: "FirstQueue", autoAck: false, consumer: consumer);
为了避免消息丢失问题,消息的确认,最好在是业务处理完再进行确认。否则会出现第三方中介出问题时,或业务处理出问题时,或刚确认好消息,业务还没处理就系统异常,导致消息未消费就丢失的问题。
这时候应答也是正常的,但数据丢失,这种情况,是这样处理的:

就两点:
1、发送信息BasicPublish方法的第三个参数:mandatory设置为true。2、定义接收的回调:BasicReturn事件。
本篇简单介绍如何使用RabbitMQ消息时,做到消息的可靠性,不丢失。
| 版权声明:本文原创发表于 博客园,作者为 路过秋天本文欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则视为侵权。 |
个人微信公众号![]() | 创业QQ群:617713515![]() | Donation(扫码支持作者):支付宝:![]() | Donation(扫码支持作者):微信:![]() |
![]() | ![]() |