ICode9

精准搜索请尝试: 精确搜索
首页 > 其他分享> 文章详细

RabbitMQ 入门系列:6、保障消息:不丢失:发送方、Rabbit存储端、接收方。

2022-08-29 10:00:37  阅读:204  来源: 互联网

标签:false 入门 RabbitMQ 消息 Rabbit var channel


系列目录

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、扩展内容:延时队列:延时队列插件及其有限的适用场景。

前言:

本篇简单介绍如何保障消息不丢失的处理方式。

1、保障消息不丢失:发送方

主要是通过消息确认或事务,来保障这个过程,下面见具体代码:

1、通过确认机制处理的代码:

复制代码
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
    {
        //超时或失败,需要处理是否重发消息。
    }
}
复制代码

2、通过事务机制处理的代码:

复制代码
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();
        //处理事务提交失败的逻辑。
    }
}
复制代码

2、保障消息不丢失:RabbitMQ端

对于RabbitMQ端的消息保障,我们人为可以处理的是,设置创建的队列或消息是否持久化。

通过创建持久化的队列或消息,可以保障消息写入硬盘,重启时仍能还原信息。

//第二个参数:是否持久化
channel.QueueDeclare("FirstQueue", true, false, false);

3、保障消息不丢失:接收方

接收方主要是通过消息确认,来指示是否收到信息。

复制代码
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);
复制代码

在以上代码中,通过指定autoAck可以自动回应收到信息,但这种自动模式,不一定能保证确认信息发回服务端。

因此,要100分百保证,需要手动回应:

复制代码
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);
复制代码

说明:

为了避免消息丢失问题,消息的确认,最好在是业务处理完再进行确认。

否则会出现第三方中介出问题时,或业务处理出问题时,或刚确认好消息,业务还没处理就系统异常,导致消息未消费就丢失的问题。

 

4、发送方:补充:还有一种情况:通过交换机发送过去,但交换机没送到指定的队列时

这时候应答也是正常的,但数据丢失,这种情况,是这样处理的:

 

就两点:

1、发送信息BasicPublish方法的第三个参数:mandatory设置为true。

2、定义接收的回调:BasicReturn事件。

总结:

本篇简单介绍如何使用RabbitMQ消息时,做到消息的可靠性,不丢失。

版权声明:本文原创发表于 博客园,作者为 路过秋天 本文欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则视为侵权。
个人微信公众号
创业QQ群:617713515
Donation(扫码支持作者):支付宝:
Donation(扫码支持作者):微信:
      转 https://www.cnblogs.com/cyq1162/p/16603046.html

标签:false,入门,RabbitMQ,消息,Rabbit,var,channel
来源: https://www.cnblogs.com/wl-blog/p/16634903.html

本站声明: 1. iCode9 技术分享网(下文简称本站)提供的所有内容,仅供技术学习、探讨和分享;
2. 关于本站的所有留言、评论、转载及引用,纯属内容发起人的个人观点,与本站观点和立场无关;
3. 关于本站的所有言论和文字,纯属内容发起人的个人观点,与本站观点和立场无关;
4. 本站文章均是网友提供,不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属;如您发现该文章侵犯了您的权益,可联系我们第一时间进行删除;
5. 本站为非盈利性的个人网站,所有内容不会用来进行牟利,也不会利用任何形式的广告来间接获益,纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

专注分享技术,共同学习,共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有