zoukankan      html  css  js  c++  java
  • .Net下RabbitMQ的使用(3) 竞争的消费者

    上一篇文章中,演示了一个发送者和一个消费者的情况。这一篇介绍一下多个消费者在同一个消息队列中获取消息的情况。

    在有些应用当中,消费端接收到消息任务需要长时间的处理,如果等上一个消息处理完成以后再取下一个数据进行处理的话,势必会有一些延迟。在消息队列中的数据也会不断增多,延迟将越来越大。当然对于一个消费进程来说,在某些情况下可以起多个线程来处理,而在这里将介绍另一种处理方式,多个消费进程的情况。而RabbitMQ在这方面进行了很好的处理和封装,使客户程序可以很方便的使用。

    python-two

    其实,在代码实现上和上一遍的例子中并没有什么不同,我们只需要运行两个Server端的程序就可以了。我们可以看到,发送的消息会平均的由两个Server中的一个来处理。不会有重复。倘若一个Server程序关闭了,那之后发送的所有消息都会在还运行着的那个Server程序上处理。这可以解决我们一些需要负载均衡的场景,而且扩展非常方便,只要在运行一个Server也就是Worker就可以了,Worker之间的状态同步都免了。

    当我们处理一个较长时间任务的时候,程序在处理过程中如果出现异常,或程序挂了导致消息没有处理成功,我们通常并不希望丢失该消息任务,而希望由其他Worker来处理或者等挂了的Worker重新起来以后再处理。同样,应对该场景,RabbitMQ也提供了简便的API方便我们处理。在RabbitMQ中,为了不让消息丢失,它提供了消息应答的概念。当消费者获取到了一个消息以后,需要给RabbitMQ服务一个应答的消息,告知服务我已经收到或正确处理了该消息。那么RabbitMQ可以放心的在队列中删除该消息。在上一篇的服务端的代码中

    channel.BasicConsume("TaskQueue", true, consumer);

    的第二个参数是true,RabbitMQ服务一旦把消息送达目标队列及认为应答了。为了演示不发送应答消息的情况,我们需要安装一个RabbitMQ的plugin:Management UI。安装这个插件比较简单。

    你可以再 http://www.rabbitmq.com/management.html 了解到详细信息。

    打开浏览器, http://localhost:55672,输入默认的用户名密码 guest/guest,进入主界面。界面上信息很多,我们只看最上面的一块:Queued Messages。

    还是启动一个发送程序和一个接受程序。把接收程序的定义消费者的代码修改为

    channel.BasicConsume("TaskQueue", false, consumer);

    然后运行两个程序,发送一个消息。由于我们在接收消息的代码中没有发送消息收到的应答包,所以在刚才监控的网页上回出现如下结果:

    image

    没有回应的消息是1个。接下来我们关闭接收程序,不修改任何代码,然后再次运行接收程序。我们发现接收程序再次收到了那个原来的消息。而浏览器上显示的状态还是没有变说明还是有一个消息没有应答,因为我们第二次运行的接收程序还是没有发应答包。关掉接收程序,修改一下代码,在处理完消息的时候我们添加如下代码:

    channel.BasicAck(ea.DeliveryTag, false);

    这时,我发现程序如我们所料,还是会收到一次那个消息包,但是监控网页的界面却变了:

    image

    然后,无论在运行多少次接收程序,都不会再收到该消息包了。这也就说明了,RabbitMQ真正认为该消息被正确处理了。

    需要注意的是,RabbitMQ对于没有发送应答包的消息没有时间的限制,也就是说没有Timeout。RabbitMQ只会在处理消息的接收程序与RabbitMQ服务端断开连接后才会重新分配该消息。如果连接没有断,但处理程序几天没有给回应包,它也不会重新发送。所以,如果在处理程序出现异常的时候,我们可以写代码将与RabbitMQ的连接断开来实现消息的重新发送(也许会发到其他负载均衡的机器上处理)。

    修改后的接收端代码如下:

        public class Worker
        {
            public void Listen()
            {
                ConnectionFactory factory = new ConnectionFactory();
                factory.HostName = "localhost";
                using (IConnection connection = factory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.QueueDeclare("TaskQueue", true, false, false, null);
                        //channel.BasicQos(500, 1, false);
                        QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
                        channel.BasicConsume("TaskQueue", false, consumer);
     
                        while (true)
                        {
                            BasicDeliverEventArgs ea =
                                (BasicDeliverEventArgs)consumer.Queue.Dequeue();
                            Console.WriteLine("Receive a Message, start to handle");
     
                            //模拟断开连接
                            return;
     
                            //模拟长时间运行
                            Thread.Sleep(60000);
                            byte[] bytes = ea.Body;
     
                            XmlSerializer xs = new XmlSerializer(typeof(RequestMessage));
                            using (MemoryStream ms = new MemoryStream(bytes))
                            {
                                RequestMessage message = (RequestMessage)xs.Deserialize(ms);
                                Console.WriteLine("Receive a Message, Id:" + message.MessageId + " Message:" + message.Message);
                            }
                            //发送应答包
                            channel.BasicAck(ea.DeliveryTag, false);
                        }
                    }
                }
            }
        }

    需要注意的是,该功能只有在point-point模式下才有如此效果,也就是一发一接的模式下。如果是发布和订阅这种broadcasting的模式下,这种配置项的结果会有一些不同,我们下一篇再说。

    完整示例代码从这里下载

  • 相关阅读:
    [linux1exe元旦特供]基于wine的IE6一键安装
    http://www.tianya.cn/publicforum/content/funinfo/1/1869814.shtml
    最全的design pattern 概述
    例解 autoconf 和 automake 生成 Makefile 文件
    根据条件分组SQL
    SQL Express自动备份
    通过asp.net 短信猫发短信
    EXT.NET GridPanel(按键事件带参的另一种写法)
    Asp.net操作IIS(调试通过)
    面向接口编程详解——编程实例(T2噬菌体)
  • 原文地址:https://www.cnblogs.com/haoxinyue/p/2703964.html
Copyright © 2011-2022 走看看