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博客安装
第四节:剖析生产者、消费者的几种消息确认机制(、事务、自动、手动) - - 博客园