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);
  
•             }
  
•         }
  

来自为知笔记(Wiz)