zoukankan      html  css  js  c++  java
  • .Net使用RabbitMQ详解

    序言

    这里原来有一句话,触犯啦天条,被阉割!!!!

    首先不去讨论我的日志组件怎么样。因为有些日志需要走网络,有的又不需要走网路,也是有性能与业务场景的多般变化在其中,就把他抛开,我们只谈消息RabbitMQ。

    那么什么是RabbitMQ,它是用来解决什么问题的,性能如何,又怎么用?我会在下面一一阐述,如有错误,不到之处,还望大家不吝赐教。

    RabbitMQ简介

    必须一提的是rabbitmq是由LShift提供的一个消息队列协议(AMQP)的开源实现,由以高性能、健壮以及可伸缩性出名的Erlang写成(因此也是继承了这些优点)。

    百度百科对RabbitMQ阐述也非常明确,建议去看下,还有amqp协议。

    RabbitMQ官网:http://www.rabbitmq.com/ 如果你要下载安装,那么必须先把Erlang语言装上。

    RabbitMQ的.net客户端,可以在nuget中输入rabbitmq轻松获得。

    RabbitMQ与其他消息队列的对比,早有仙人给写出来。 Message Queue Shootout

    这篇文章中的测试案例为:1百万条1k的消息,每秒种的收发情况如下图。

    如果你安装好啦,rabbitmq,他会提供一个操作监控页面,页面如下,他几乎提供啦,对rabbitmq的所有操作,与监控,所以,你装上后,自己多看看,多操作下。

     

    RabbitMQ中的一些名词阐述与消息从投递到消费的整个过程

    从上图的标题中可以看到一些陌生的英文单词,让我们感觉一无所知,更无从操作,那么我给大家弄啦一个图片大家可以看下,或许对您理解这些新鲜的单词有所帮助。

    看过这些名词,之后,或许你还毫无头绪,那么我把消息从生产到消费的整个流程给大家说一下,或许会更深入一点,其中Exchange,与Queue都是可以设置相关属性,队列的持久化,交换器类型制定。

    Note:首先这个过程走分三个部分,1、客户端(生产消息队列),2、RabbitMQ服务端(负责路由规则的绑定与消息的分发),3、客户端(消费消息队列中的消息)

    Note:由图可以看出,一个消息可以走一次网络却被分发到不同的消息队列中,然后被多个的客户端消费,那么这个过程就是RabbitMQ的核心机制,RabbitMQ的路由类型与消费模式。

    RabbitMQ中Exchange的类型

    类型有4种,direct,fanout,topic,headers。其中headers不常用,本篇不做介绍,其他三种类型,会做详细介绍。

    那么这些类型是什么意思呢?就是Exchange与队列进行绑定后,消息根据exchang的类型,按照不同的绑定规则分发消息到消息队列中,可以是一个消息被分发给多个消息队列,也可以是一个消息分发到一个消息队列。具体请看下文。

    介绍之初还要说下RoutingKey,这是个什么玩意呢?他是exchange与消息队列绑定中的一个标识。有些路由类型会按照标识对应消息队列,有些路由类型忽略routingkey。具体看下文。

    1、Exchange类型direct

    他是根据交换器名称与routingkey来找队列的。

    Note:消息从client发出,传送给交换器ChangeA,RoutingKey为routingkey.ZLH,那么不管你发送给Queue1,还是Queue2一个消息都会保存在Queue1,Queue2,Queue3,三个队列中。这就是交换器的direct类型的路由规则。只要找到路由器与routingkey绑定的队列,那么他有多少队列,他就分发给多少队列。

    2、Exchange类型fanout

    这个类型忽略Routingkey,他为广播模式。

    Note:消息从客户端发出,只要queue与exchange有绑定,那么他不管你的Routingkey是什么他都会将消息分发给所有与该exchang绑定的队列中。

    3、Exchange类型topic

    这个类型的路由规则如果你掌握啦,那是相当的好用,与灵活。他是根据RoutingKey的设置,来做匹配的,其中这里还有两个通配符为:

    *,代表任意的一个词。例如topic.zlh.*,他能够匹配到,topic.zlh.one ,topic.zlh.two ,topic.zlh.abc, ....

    #,代表任意多个词。例如topic.#,他能够匹配到,topic.zlh.one ,topic.zlh.two ,topic.zlh.abc, ....

    Note:这个图看上去很乱,但是他是根据匹配符做匹配的,这里我建议你自己做下消息队列的具体操作。

    具体操作如下

    复制代码
     public static void Producer(int value)
            {
                try
                {
                    var qName = "lhtest1";
                    var exchangeName = "fanoutchange1";
                    var exchangeType = "fanout";//topic、fanout
                    var routingKey = "*";
                    var uri = new Uri("amqp://192.168.10.121:5672/");
                    var factory = new ConnectionFactory
                    {
                        UserName = "123",
                        Password = "123",
                        RequestedHeartbeat = 0,
                        Endpoint = new AmqpTcpEndpoint(uri)
                    };
                    using (var connection = factory.CreateConnection())
                    {
                        using (var channel = connection.CreateModel())
                        {
                            //设置交换器的类型
                            channel.ExchangeDeclare(exchangeName, exchangeType);
                            //声明一个队列,设置队列是否持久化,排他性,与自动删除
                            channel.QueueDeclare(qName, true, false, false, null);
                            //绑定消息队列,交换器,routingkey
                            channel.QueueBind(qName, exchangeName, routingKey);
                            var properties = channel.CreateBasicProperties();
                            //队列持久化
                            properties.Persistent = true;
                            var m = new QMessage(DateTime.Now, value+"");
                            var body = Encoding.UTF8.GetBytes(DoJson.ModelToJson<QMessage>(m));
                            //发送信息
                            channel.BasicPublish(exchangeName, routingKey, properties, body);
                        }
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
            }
    复制代码

    消息队列的消费与消息确认Ack

    1、消息队列的消费

    Note:如果一个消息队列中有大量消息等待操作时,我们可以用多个客户端来处理消息,这里的分发机制是采用负载均衡算法中的轮询。第一个消息给A,下一个消息给B,下下一个消息给A,下下下一个消息给B......以此类推。

    2、为啦保证消息的安全性,保证此消息被正确处理后才能在服务端的消息队列中删除。那么rabbitmq提供啦ack应答机制,来实现这一功能。

    ack应答有两种方式:1、自动应答,2、手动应答。具体实现如下。

    复制代码
     public static void Consumer()
            {
                try
                {
                    var qName = "lhtest1";
                    var exchangeName = "fanoutchange1";
                    var exchangeType = "fanout";//topic、fanout
                    var routingKey = "*";
                    var uri = new Uri("amqp://192.168.10.121:5672/");
                    var factory = new ConnectionFactory
                    {
                        UserName = "123",
                        Password = "123",
                        RequestedHeartbeat = 0,
                        Endpoint = new AmqpTcpEndpoint(uri)
                    };
                    using (var connection = factory.CreateConnection())
                    {
                        using (var channel = connection.CreateModel())
                        {
                            channel.ExchangeDeclare(exchangeName, exchangeType);
                            channel.QueueDeclare(qName, true, false, false, null);
                            channel.QueueBind(qName, exchangeName, routingKey);
                            //定义这个队列的消费者
                            QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
                            //false为手动应答,true为自动应答
                            channel.BasicConsume(qName, false, consumer);
                            while (true)
                            {
                                BasicDeliverEventArgs ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();                           
                                byte[] bytes = ea.Body;
                                var messageStr = Encoding.UTF8.GetString(bytes);
                                var message = DoJson.JsonToModel<QMessage>(messageStr);
                                Console.WriteLine("Receive a Message, DateTime:" + message.DateTime.ToString("yyyy-MM-dd HH:mm:ss") + " Title:" + message.Title);
                                //如果是自动应答,下下面这句代码不用写啦。
                                if ((Convert.ToInt32(message.Title) % 2) == 1)
                                {
                                    channel.BasicAck(ea.DeliveryTag, false);
                                }
                            }
                        }
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
            }
    复制代码

    转:http://www.cnblogs.com/knowledgesea/p/5296008.html

  • 相关阅读:
    MacOS的多重启动工具
    exchange 2003配置ASSP 反垃圾邮件
    VMWARE workstation 9 收缩虚拟硬盘
    [AX2012]在SSRS报表中获取从Menuitem传入的记录
    [AX2012]代码更改默认财务维度
    js使用模板快速填充数据
    CI框架--事务
    $().each 和表单事件的坑
    基于layerpage 前后端异步分页
    SQL 语句格式
  • 原文地址:https://www.cnblogs.com/soundcode/p/8515623.html
Copyright © 2011-2022 走看看