栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > .Net

.net RabbitMQ之CreateBasicProperties:Lazy Queue,Persistent Message

.Net 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

.net RabbitMQ之CreateBasicProperties:Lazy Queue,Persistent Message

注:以下程序使用C#实现

我们在使用RabbitMQ发送消息到队列之后,在某些场景下,我们希望当消息没有被消费时,如果系统发生故障停止或者我们手动重启服务之后,消息仍然存在队列之中,下面我们看一下普通的使用场景:

var channel=connection.CreateModel();
channel.QueueDeclare("mytest",false,false,false,null);
channel.QueueBind("mytest",string.Empty,"",null);
for(var i=0;i<10;i++){
channel.BasicPublish(string.Empty,"mytest",null,Encoding.UTF8.GetBytes("你好"));
}

上面代码是我们往mytest队列中塞了10条数据进去,这10条消息并没有被消费,但是当我们重启RabbitMQ之后,会发现,mytest队列中的消息变为NaN了,代表消息已经丢失了,在某些时候,由于重启服务而丢失消息,是我们不希望看到的

解决办法:

在程序中使用CreateBasicProperties方法,通过该方法创建一个属性实例,给实例的Persistent字段赋值为true即可

var channel=connection.CreateModel();
channel.QueueDeclare("mytest",false,false,false,null);
channel.QueueBind("mytest",string.Empty,"",null);
var properties=channel.CreateBasicProperties();
properties.Persistent=true;
for(var i=0;i<10;i++){
channel.BasicPublish(string.Empty,"mytest",properties,Encoding.UTF8.GetBytes("你好"));
}

下面介绍一下RabbitMQ中的Lazy Queue用法。注:LazyQueue就是将消息写入磁盘,只有用到的时候,才会写入内存

如果我们采用Lazy Queue+InMemory也就是配置了LazyQueue并且properties.Persistent=false;

就是消息不写入内存,写入磁盘

var channel=connection.CreateModel();
//使用方法
channel.QueueDeclare("mytest",false,false,false,new Dictionary{
{"x-queue-mode","lazy"}//设置为懒队列
});
channel.QueueBind("mytest",string.Empty,"",null);

for(var i=0;i<100000;i++){
//注意!以下的msg数据非常大,每个msg高达百兆
var msg=string.Format("{0}测试",string.join(",",Enumrable.Range(1,100000)));
channel.BasicPublish(string.Empty,"mytest",null,Encoding.UTF8.GetBytes(msg));
}

这样发送消息,我们会发现,内存的大小几乎没有改变,这说明Lazy确实不会将消息写入内存,而是临时存储在磁盘上,但是这种写法会有问题,就是重启服务之后,消息会丢失。

为了不丢失消息,我们需要采用这种写法 Lazy Queue+Disk

var channel=connection.CreateModel();
//添加下面两句
var properties=channel.BasicCreateProperties();
properties.Persistent=true;
//使用方法
channel.QueueDeclare("mytest",false,false,false,new Dictionary{
{"x-queue-mode","lazy"}//设置为懒队列
});
channel.QueueBind("mytest",string.Empty,"",null);

for(var i=0;i<100000;i++){
//注意!以下的msg数据非常大,每个msg高达百兆
var msg=string.Format("{0}测试",string.join(",",Enumrable.Range(1,100000)));
channel.BasicPublish(string.Empty,"mytest",null,Encoding.UTF8.GetBytes(msg));
}

这样,内存既不会暴增,也能将消息持久化

如果是Default Queue+InMemory方法

var channel=connection.CreateModel();

//使用方法
channel.QueueDeclare("mytest",false,false,false,null);
channel.QueueBind("mytest",string.Empty,"",null);

for(var i=0;i<100000;i++){
//注意!以下的msg数据非常大,每个msg高达百兆
var msg=string.Format("{0}测试",string.join(",",Enumrable.Range(1,100000)));
channel.BasicPublish(string.Empty,"mytest",null,Encoding.UTF8.GetBytes(msg));
}

这种方式内存会暴增,并且消息不会持久化到磁盘

我们可以采用Default Queue+Disk方式,将消息持久化到磁盘

var channel=connection.CreateModel();
//添加下面两句
var properties=channel.BasicCreateProperties();
properties.Persistent=true;
//使用方法
channel.QueueDeclare("mytest",false,false,false,null);
channel.QueueBind("mytest",string.Empty,"",null);

for(var i=0;i<100000;i++){
//注意!以下的msg数据非常大,每个msg高达百兆
var msg=string.Format("{0}测试",string.join(",",Enumrable.Range(1,100000)));
channel.BasicPublish(string.Empty,"mytest",null,Encoding.UTF8.GetBytes(msg));
}

但是需要注意的是,这种方式,内存也会暴增

综上所述,如果我们的消息不需要相应非常快,并且硬件内存不足时,可以采用Lazy Queue+Disk方式,如果时其它情况,则需另行讨论

转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/910949.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号