当前位置: 首页 > news >正文

无锡公司网站建设服务免费生成短链接

无锡公司网站建设服务,免费生成短链接,郑州团购网站建设,专业设计vi背景 最近遇到了一个问题,在使用rabbitmq的时候出现了丢消息、消息重复消费等一系列的问题,使用的是.net框架,背景是高并发压力下的mq消费,按理说即使队列中堆了几百条消息,我客户端可以同处理5个消息。 原因是多线程…

背景

最近遇到了一个问题,在使用rabbitmq的时候出现了丢消息、消息重复消费等一系列的问题,使用的是.net框架,背景是高并发压力下的mq消费,按理说即使队列中堆了几百条消息,我客户端可以同处理5个消息。

原因是多线程同时处理时导致的内存混乱。

官方文档已经解释的很全面了:https://www.rabbitmq.com/dotnet-api-guide.html

一个简易的单线程消费者

注意如下代码,这只是一个简易的单线程同步的消费者;
每次消费1条消息,消息消费完进行手动ack;

Task.Run(() =>{AutoResetEvent autoResetEvent = new AutoResetEvent(false);ConnectionFactory factory = new ConnectionFactory();// "guest"/"guest" by default, limited to localhost connectionsfactory.UserName = user;factory.Password = pass;factory.VirtualHost = vhost;factory.HostName = hostName;// this name will be shared by all connections instantiated by// this factoryfactory.ClientProvidedName = "app:audit component:event-consumer";IConnection conn = factory.CreateConnection();using (IModel channel = conn .CreateModel()){channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);channel.QueueDeclare(queueName, false, false, false, null);channel.QueueBind(queueName, exchangeName, routingKey, null);consumer.Received += (ch, ea) =>{var body = ea.Body.ToArray();// copy or deserialise the payload// and process the message// ...channel.BasicAck(ea.DeliveryTag, false);};channel.BasicConsume(queue: "my-queue",autoAck: false,consumer: consumer);}ConsoleUtil.WriteLine("mq started");autoResetEvent.WaitOne();ConsoleUtil.WriteLine("mq shutdown");}
});

批量消费

好的,那么我现在想要同时消费5条消息,想要达到并行的效果,需要如何改代码呢?看下面的改动:

改动1:

先看两个概念

  1. prefetchCount(预取计数):

    • prefetchCount 是一个用来限制每个消费者一次性从队列中获取的消息数量的参数。
    • 当你有多个消费者同时连接到同一个队列时,RabbitMQ 可以将消息均匀地分发给这些消费者。
    • 通过设置 prefetchCount,你可以告诉 RabbitMQ 每个消费者一次最多获取多少条消息。
    • 这个参数的目的是确保消息在被消费者处理之前不会全部放到内存中,从而提高系统的稳定性和性能。它有助于避免 一个消费者获取了太多消息而导致其他消费者无法获取任何消息的情况。
  2. concurrentConsumers(并发消费者):

    • concurrentConsumers 是指在同一队列上允许多少个并发消费者。
    • 每个并发消费者都会独立地处理消息,这有助于提高系统的处理能力和吞吐量。
    • 通过增加 concurrentConsumers 数量,你可以增加并发处理消息的能力。
    • 注意,这个参数不同于 prefetchCount,它控制的是同时运行的消费者的数量,而不是单个消费者一次性获取的消息数量。
// 参数1:prefetchSize:可接收消息的大小,如果设置为0,那么表示对消息本身的大小不限制
// 参数2:prefetchCount:处理消息最大的数量。相当于消费者能一次接受的队列大小
// 参数3:global:是不是针对整个 Connection 的,因为一个 Connection 可以有多个 Channel
// global=false:针对的是这个 Channel
// global=ture: 针对的是这个 Connection
channel.BasicQos(0, 5, false);
factory.ConsumerDispatchConcurrency = 5;

好的,这时候我配置了同时处理5条消息,看起来没问题了,但是官网文档有这样一句话:

IModel instance usage by more than one thread simultaneously should be avoided. Application code should maintain a clear notion of thread ownership for IModel instances.

This is a hard requirement for publishers: sharing a channel (an IModel instance) for concurrent publishing will lead to incorrect frame interleaving at the protocol level. Channel instances must not be shared by threads that publish on them.

If more than one thread needs to access a particular IModel instances, the application should enforce mutual exclusion. One way of achieving this is for all users of an IModel to lock the instance itself:

大概意思就是应该避免多个线程同时使用IModel实例,也就是channel对象,如果这么做的后果就是高负载情况下导致内存混乱,有可能你的线程1消费到了线程5本该消费的消息,这听起来后果是很严重的,那么我们应该怎么改动呢?官网也给方案了,就是给channel对象加锁,看下面的代码改动:

改动2

consumer.Received += (ch, ea) =>{	var body = ea.Body.ToArray();// copy or deserialise the payload// and process the message// ...lock (channel){channel.BasicAck(ea.DeliveryTag, false);}
};lock (channel){channel.BasicConsume(queue: "my-queue",autoAck: false,consumer: consumer);
}

异步支持

新增一个配置:

factory.DispatchConsumersAsync = true;

然后修改消费者:

var consumer = new AsyncEventingBasicConsumer(channel);consumer.Received += async (model, ea) =>
{await Task.Run(() =>{var body = ea.Body.ToArray();// copy or deserialise the payload// and process the message// ...lock (channel){channel.BasicAck(ea.DeliveryTag, false);}});
};
http://www.zhongyajixie.com/news/16272.html

相关文章:

  • 有没有专门做尾料回收的网站专业做灰色关键词排名
  • 网站建设官网免费模板搜索网站有哪些
  • 做网站提升公司形象站长是什么职位
  • nodejs网站开发关键词排名优化流程
  • 网站建设行业新闻销售课程视频免费
  • 怎么搭建一个自己的网站系统优化大师下载
  • 网站各个阶段推广友情下载网站
  • 西安高端品牌网站建设广州网络推广公司有哪些
  • 合肥专业网站建设手机网站百度关键词排名查询
  • 顺义企业网站建站公司站长工具seo综合查询是什么
  • 中山建设网站首页东莞seo快速排名
  • 如何查看网站的更新频率网站设计公司苏州
  • wordpress 封ip基本seo
  • 上海风险地区划分最新查询购买seo关键词排名优化官网
  • 好的企业型网站模板百度163黄页关键词挖掘
  • 深圳市公司网站建设公司电子商务说白了就是干什么的
  • 网站添加flash网络营销推广方案
  • 聊城商城网站建设怎样制作网站
  • 免费网站统计windows永久禁止更新
  • 网站做百度收录的意义如何在百度提交自己的网站
  • 网站建设培训速成电商代运营公司100强
  • 秦皇岛网站制作与网站建设迅雷磁力链bt磁力天堂
  • 哪里可以做产品购物网站白酒营销策划方案
  • 郑州哪些公司做网站比较好企业内训
  • 上海微信小程序网站建设网站交换链接的常见形式
  • 无锡论坛网站制作上海网络推广渠道
  • 做网站用什么技术好百度seo快排软件
  • 做旅游网站的意义淘宝搜索关键词查询工具
  • 亚马逊备案网站建设河南郑州最新消息
  • 网站建设多少钱一年交换友链是什么意思