欢迎您访问安装NuGet包.2生产者-确认机制(1).!

安装NuGet包.2生产者-确认机制(1).

更新时间:2022-11-26 07:09:18作者:佚名

1、安装NuGet包.

2 生产者-确认机制

(1). 含义:就是应答模式,生产者发送一条消息之后接收频道信息失败 请重试,服务器做了个响应,表示收到了。

(2). 特点:异步模式,在响应之前,可以继续发送消息,单条消息、批量消息均可继续发送。

(3). 核心代码:单条消息确认:.()

批量消息确认:.()

大致流程:.(); 开启确认模式→发送消息→提供一个回执方法(); 返回一个bool 值

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
var factory = new ConnectionFactory();
factory.HostName = "127.0.0.1";//RabbitMQ服务器地址
factory.DispatchConsumersAsync = true;//支持异步发送消息
string exchangeName = "exchange1";//交换机的名字
string eventName = "myEvent";// routingKey的值
using var conn = factory.CreateConnection();
while (true)
{
    string msg = DateTime.Now.TimeOfDay.ToString();//待发送消息
    using (var channel = conn.CreateModel())//创建信道
    {
        try
        {
            var properties = channel.CreateBasicProperties();
            properties.DeliveryMode = 2;//1非持久化、2是持久化
            //交换机 Name:交换机名称。
            // Type: 交换机类型——direct、topic、fanout、headers、sharding
            //Durable:消息代理重启后,交换机是否还存在。
            //Auto-delete :当所有与之绑定的消息队列都完成了对此交换机的使用后,删掉它。
            //Arguments:依赖代理本身。
            channel.ExchangeDeclare(exchange: exchangeName, type: "direct", true, false);//声明交换机
            byte[] body = Encoding.UTF8.GetBytes(msg);
            channel.ConfirmSelect();//开启消息确认模式
            //发布消息    
            //exchange:交换机名称
            //mandatory为true时,表示如果消息没有被正确路由,消息将退回消息的生产者 如果设置为false,那么broker端自动删除该消息。
            //routingKey:路由键
            //props:消息属性字段,比如消息头部信息等等
            //body:消息主体部分
            channel.BasicPublish(exchange: exchangeName, routingKey: eventName,
            mandatory: true, basicProperties: properties, body: body);
            Console.WriteLine("发布了消息:" + msg);
            /*首先开启Confirm模式,通知消息生产者成功推送到RabbitMQ中*/
            if (channel.WaitForConfirms())  //单条消息确认
            {
                //表示消息发送成功(已经存入队列)
                Console.WriteLine($"【{msg}】成功发送到RabbitMQ!");
            }
            else
            {
                Console.WriteLine($"【{msg}】发送到RabbitMQ失败!");
            }
            //channel.WaitForConfirmsOrDie();//如果所有消息发送成功 就正常执行, 如果有消息发送失败;就抛出异常;
        }

接收频道信息失败 请重试

catch (Exception ex) { //表示消息发送失败 Console.WriteLine($"【{msg}】发送到RabbitMQ失败!"); } } Thread.Sleep(1000); }

运行结果:

接收频道信息失败 请重试

2. TX事务模式

(1). 含义:基于AMPQ协议;可以让信道设置成一个带事务的信道,分为三步:开启事务、提交事务、事务回滚

(2). 特点:同步模式接收频道信息失败 请重试,在事务提交之前不能继续发送消息,该模式相比模式效率差一点。

(3). 核心代码:.(); 开启一个事务

.(); 提交事务, 这一步成功后,消息才真正的写入队列

.(); 事务回滚

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
var factory = new ConnectionFactory();
factory.HostName = "127.0.0.1";
factory.DispatchConsumersAsync = true;
string exchangeName = "exchange1";
string eventName = "myEvent";
using var conn = factory.CreateConnection();
while (true)
{
    string msg = DateTime.Now.TimeOfDay.ToString();
    using (var channel = conn.CreateModel())
    {
        try
        {
            var properties = channel.CreateBasicProperties();
            properties.DeliveryMode = 2;
            channel.ExchangeDeclare(exchange: exchangeName, type: "direct", true, false);
            byte[] body = Encoding.UTF8.GetBytes(msg);
            channel.TxSelect(); //开启事务
            channel.BasicPublish(exchange: exchangeName, routingKey: eventName,
            mandatory: true, basicProperties: properties, body: body);
            Console.WriteLine($"【{msg}】成功发送到RabbitMQ!");
            channel.TxCommit(); //只有事务提交成功以后,才会真正的写入到队列里面去
        }
        catch (Exception ex)
        {
            //表示消息发送失败
            Console.WriteLine($"【{msg}】发送到RabbitMQ失败!");
            channel.TxRollback();
        }
    }
    Thread.Sleep(1000);
}

3、消费者(手动确认)

(1) 含义:消费者消费一条,回执给一条消息, 只删除当前这一条消息;相当于是一条消费了,删除一条消息,性能稍微低一些;

(2) 特点:消费1条应答一次,可以告诉消费成功or失败,消费成功,服务器删除该条消息,消费失败,可以删除也可以重新写入。

(3) 核心代码:: false,表示不自动确认

然后:.(: ea., : false); 表示消费成功

.(: ea., : false); 表示消费失败, 可以配置:: true:重新写入到队列里去; false: 删除消息

using RabbitMQ.Client;

接收频道信息失败 请重试

using RabbitMQ.Client.Events; using System.Text; var factory = new ConnectionFactory(); factory.HostName = "127.0.0.1";//服务器地址 factory.DispatchConsumersAsync = true;//支持异步接受 string exchangeName = "exchange1";//交换机的名称 string eventName = "myEvent";//路由键 using var conn = factory.CreateConnection(); using var channel = conn.CreateModel();//创建信道 string queueName = "queue1";//队列名称 //声明了交换机 channel.ExchangeDeclare(exchange: exchangeName, type: "direct",true,false); //声明一个队列 //queuename: 队列的名称 //durable 是否持久化。true消息会持久化到本地,保证重启服务后消息不会丢失 //exclusive :表示独占方式,设置为true 在某些情景下有必要,例如:顺序消费。表示只有一个channel可以去监听,其他channel都不能够监听。目的就是为了保证顺序消费。 //autoDelete:队列如果与Exchange未绑定,则自动删除 //arguments:扩展参数 channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null); //绑定队列 channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: eventName); //创建一个消费者 var consumer = new AsyncEventingBasicConsumer(channel); consumer.Received += Consumer_Received; //注册消费者订阅 //autoAck 是否自动确认消息, true自动确认,false 不自动要手动调用, 建立设置为false channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer); Console.ReadLine(); async Task Consumer_Received(object sender, BasicDeliverEventArgs args) { try { var bytes = args.Body.ToArray(); string msg = Encoding.UTF8.GetString(bytes); Console.WriteLine(DateTime.Now + "收到了消息" + msg); //DeliveryTag: 唯一的编号 //multiple:是否批量确认.true:将一次性ack所有小于deliveryTag的消息。 channel.BasicAck(args.DeliveryTag, multiple: false);手动确认 await Task.Delay(800); } catch (Exception ex) { //异常重试 //DeliveryTag: 唯一的编号 //requeue:如果 requeue 参数设置为 true,则 RabbitMQ 会重新将这条消息存入队列,以便发送给下一个订阅的消费者; 如果 requeue 参数设置为 false,则 RabbitMQ 立即会还把消息从队列中移除,而不会把它发送给新的消费者。 channel.BasicReject(args.DeliveryTag, true); Console.WriteLine("处理收到的消息出错" + ex); } }

运行结果

接收频道信息失败 请重试

参考

【安装详细教程】_慕之寒的博客-CSDN博客安装

第四节:剖析生产者、消费者的几种消息确认机制(、事务、自动、手动) - - 博客园

为您推荐

红星家具集团有限公司亿元从融创中国控股有限公司大股东

而红星家具集团与红星美凯龙控股集团(以下简称“红星控股”)属于兄弟公司的关系,可见,该交易和红星控股旗下的上市公司红星美凯龙(.此次交易并非红星家具集团及其相关公司首次和金科产生交集。此后,红星美凯龙2017年从金科股份买下了重庆江北家居商场项目;2018年签约购入重庆照母山家居商场项目,并拟于今年开工;2019年双方就济南家居商场项目达成委托经营管理合作。

2022-11-25 22:11

珠海旅游建议游玩时间:几个情侣北路就可以结婚了

珠海是一座海滨城市,旅游资源丰富。来到珠海旅游,一定要去看看港珠澳大桥。珠海旅游攻略农科奇观>圆明新园>情侣路>珠海渔女>澳门环岛游珠海旅游必玩的14个景点全攻略飞沙滩海滨旅游区位于珠海西部高栏岛的东南部,被评为珠海十景。金海滩位于珠海市西区三灶岛长沙湾,是珠海市十大旅游景区之一。

2022-11-25 15:12

章子怡汪峰秀恩爱沙滩全婐照疯传扒裤摸私处不堪入目

据了解,自从汪章恋曝光后,网友们大都不看好,有些网友还扒出章子怡早前的沙滩全婐照片讽刺汪峰。在这些照片中,章子怡大方扒裤,被前男友摸私处,毫不在意。沙滩全婐照疯传扒裤摸私处不堪入目”,照片中,章子怡身穿黑色上衣搭配围脖,将肚子遮住,面带笑容与汪峰站在一起。沙滩全婐照疯传扒裤摸私处不堪入目沙滩全婐照疯传扒裤摸私处不堪入目Nevo到加勒比海度假享受二人世界,章子怡被发现在沙滩大解放,赤裸上身露两点晒太阳,Vivi沙滩全婐照疯传扒裤摸私处不堪入目

2022-11-25 14:12

20位激活码,一共有26的20次方种可能性!

激活码顶多几百万个,顶多了顶顶多了!以下是所谓的激活码生成器(其实就是个暴力破解器)的核心代码://随机发生器初始化如上就是穹之扉的激活码生成器了。这样做能找到激活码吗,当然能!天帝为寻回爱女,强行开启天门,却在无意中,引发妖人乱世。华胥国主以上古神器阵列强行关闭天门,是福是祸?上古神器太一轮再现穹之扉,又将引发怎样的故事司空宇手臂上的纹路,与颛顼手上纹路是否存在关联?

2022-11-25 12:07

彭昱畅资源多是谁在捧他?个人资料及作品彭昱畅个人资料

那么彭昱畅个人资料及作品,彭昱畅资源多是谁在捧他?彭昱畅个人资料及作品彭昱畅个人资料及作品,这个人物是来自于我们的台湾地区的,而且他出演过许许多多的作品,其中代表作品有《闪光少女》,这些非常出名评分非常高的作品。彭昱畅资源多是谁在捧他?

2022-11-25 09:20

《狄仁杰之通天帝国》刘嘉玲李冰冰邓超领衔主演谁最美

刘嘉玲凭借《狄仁杰之通天帝国》获得第30届香港电影金像奖影后,这是刘嘉玲首次获此殊荣,此前刘嘉玲已经凭借《阿飞正传》等片已经提名五次。《狄仁杰之通天帝国》五位主演刘德华刘嘉玲邓超李冰冰和梁家辉在片中造型高达40多款,全片共制作了近五千套服装。当然,刘嘉玲的演技是配得上影后称号的;之后新出的续集中,刘德华被换了,刘嘉玲仍然保留,也可见武则天这个人物的重要性,同时刘嘉玲对饰演武则天的不同替代。

2022-11-25 08:18

加载中...