幂等性
MQ消费者的幂等性的解决一般使用全局ID或者写个唯一标识比如时间戳或者UUID或者订单消费者消费MQ中的消息也可利用MQ的该id来判断，或者可按自己的规则生成一个全局唯一id，每次消费消息时用该id先判断该消息是否已消费过
在海量订单生成的业务高峰期，生产端有可能就会重复发生了消息，这时候消费端就要实现幂等性，这就意味着我们的消息永远不会被消费多次，即使我们收到了一样的消息。业界主流的幂等性有两种操作:a.唯一ID+指纹码机制,利用数据库主键去重, b.利用redis的原子性去实现
| 场景 | 解决思路 | 描述 |
|---|---|---|
| 消费端的幂等性保障 | 唯一ID+指纹码机制 | 指纹码:我们的一些规则或者时间戳加别的服务给到的唯一信息码,它并不一定是我们系统生成的，基本都是由我们的业务规则拼接而来，但是一定要保证唯一性，然后就利用查询语句进行判断这个id是否存在数据库中,优势就是实现简单就一个拼接，然后查询判断是否重复；劣势就是在高并发时，如果是单个数据库就会有写入性能瓶颈当然也可以采用分库分表提升性能，但也不是我们最推荐的方式 |
| Redis原子性 | 利用redis执行setnx命令，天然具有幂等性。从而实现不重复消费 |
优先级队列
场景
- 优先处理大客户订单（不同企业规模(500人,200人,100人)在京东下单购买办公用品，优先给人多的企业发货）
注意：设置队列的最大优先级最大可设置0~255 官网推荐设置0~9 如果设置太高会比较占用cup和内存
生产者代码
using RabbitMQ.Client;
using System.Reflection;
using System.Text;
using System.Threading.Channels;
namespace _01.Rabbitmq.Producer
{
public class Program
{
private static string queueName = "hello";
static void Main(string[] args)
{
// 创建一个连接工厂
ConnectionFactory factory = new ConnectionFactory
{
HostName = "localhost",
UserName = "guest",
Password = "guest",
};
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
/*
*生成一个队列
*1.队列名称
*2.队列里面的消息是否持久化默认消息存储在内存中
*3.该队列是否只供一个消费者进行消费是否进行共享true可以多个消费者消费
*4.是否自动删除最后一个消费者端开连接以后该队列是否自动删除true自动删除*
*5.其他参数
*/
var arguments = new Dictionary<string, object>();
arguments.Add("x-max-priority", 9);
channel.QueueDeclare("hello", true, false, false, arguments);//创建一个名称为hello的消息队列
//channel.QueueDeclare("hello", false, false, false, null);//创建一个名称为hello的消息队列
for (int i = 0; i < 20; i++)
{
string message = i + "Hello World"; //控制台传递的消息内容
var body = Encoding.UTF8.GetBytes(message);
/*
* 发送一个消息
* 1.发送到那个交换机
* 2.路由的 key 是哪个
* 3.其他的参数信息
* 4.发送消息的消息体
*/
if (i == 5)
{
var basicProperties = channel.CreateBasicProperties();
basicProperties.Priority = 5;
channel.BasicPublish(exchange: "", "hello", basicProperties, body); //开始传递
}
else
{
channel.BasicPublish(exchange:"", "hello", null, body); //开始传递
}
Console.WriteLine("已发送： {0}", message);
}
Console.ReadKey();
}
}
}
消费者者代码
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
namespace _01.Rabbitmq.Consumer
{
public class Program
{
private static string queueName = "hello";
static void Main(string[] args)
{
// 创建一个连接工厂
ConnectionFactory factory = new ConnectionFactory
{
HostName = "localhost",
UserName = "guest",
Password = "guest",
};
// publisher／consumer和broker之间的TCP连接
using var connection = factory.CreateConnection();
// Channel作为轻量级的Connection极大减少了操作系统建立TCPconnection的开销
using var channel = connection.CreateModel();
var arguments = new Dictionary<string, object>();
arguments.Add("x-max-priority",9);
// 创建队列/交换机（如队列/交换机已存在的情况可不用再次创建/此创建为：确保先开启消费者，生产者未创建队列/交换机而引发报错）
channel.QueueDeclare("hello", true, false, false, arguments);
//channel.QueueDeclare("hello", false, false, false, null);
// 事件对象
var consumer = new EventingBasicConsumer(channel);
/*
* 消费者消费消息
* 1.消费哪个队列
* 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
* 3.消费者未成功消费的回调
*/
channel.BasicConsume("hello", true, consumer);
// 接收消息回调
consumer.Received += (sender, e) =>
{
var body = e.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
Console.WriteLine("已接收： {0}", message);
};
Console.ReadKey();
}
}
}
测试效果
惰性队列
场景
长时间不能消费消息造成堆积
配置
var arguments = new Dictionary<string, object>();
arguments.("x-queue-mode", "lazy");
channel.QueueDeclare("hello", true, false, false, arguments);
性能对比
在发送1百万条消息，每条消息大概占1KB的情况下，普通队列占用内存是1.2GB，而惰性队列仅仅占用1.5MB，但是消费惰性队列的消息就比较慢，因为要先从磁盘读取在加载到内存中，根据场景选择