安装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博客安装

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

相关推荐

猜你喜欢

大家正在看

换一换