RabbitMQ的使用

1、RabbitMQ介绍

1.1什么是队列？

消息队列（Message Queue，简称MQ），从字面意思上看，本质是个队列，FIFO先入先出，只不过队列中存放的内容是message而已。

其主要用途：不同进程Process/线程Thread之间通信。

1.2RabbitMQ简介

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。 AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

官方地址：https://www.rabbitmq.com

1.3.搭建RabbitMQ环境

Windows下安装

安装Erlang

下载：http://www.erlang.org/download/otp_win64_17.3.exe

 安装：RabbitMQ

下载：http://www.rabbitmq.com/download.html

1.4基本对象使用介绍

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。ConnectionFactory为Connection的制造工厂。 Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

Queue

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。

 

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。

 

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。 

 

Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。 这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

另外pub message是没有ack的。

Message durability

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

Prefetch count

前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

 

Exchange

在上一节我们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。 

 

Exchange是按照什么逻辑将消息路由到Queue的？这个将在Binding一节介绍。 RabbitMQ中的Exchange有四种类型，不同的类型有着不同的路由策略，这将在Exchange Types一节介绍。

routing key

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。 在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。 RabbitMQ为routing key设定的长度限制为255 bytes。

Binding

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

 

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。 binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

Exchange Types

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义），下面分别进行介绍。

fanout

fanout类型的Exchange路由规则非常简单，消息广播，无路由交换器，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中，。

 

上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

direct

direct类型的Exchange路由规则也很简单，单播的模式，明确的路由规则，它会把消息路由到那些binding key与routing key完全匹配的Queue中。 

 

以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

topic

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

• routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
• binding key与routing key一样也是句点号“. ”分隔的字符串
• binding key中可以存在两种特殊字符“*”与“#”，用于做模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）

 

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。

headers

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。 在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。（目前没有使用到）

RPC

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。 但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。（目前没有使用到）

 

RabbitMQ  中实现RPC的机制是：

1)    客户端发送请求（消息）时，在消息的属性（MessageProperties ，在AMQP 协议中定义了14中properties ，这些属性会随着消息一起发送）中设置两个值replyTo （一个Queue 名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue 中）和correlationId （此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）

2)    服务器端收到消息并处理

3)    服务器端处理完消息后，将生成一条应答消息到replyTo 指定的Queue ，同时带上correlationId 属性

4)    客户端之前已订阅replyTo 指定的Queue ，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理

2、同类产品对比

1.1从社区活跃度

按照RabbitMQ 、ActiveMQ 、ZeroMQ 三者中，综合来看，RabbitMQ 是首选。

1.2持久化消息比较

ZeroMQ 不支持，ActiveMQ 和RabbitMQ 都支持。持久化消息主要是指我们机器在不可抗力因素等情况下挂掉了，消息不会丢失的机制。

1.3综合技术实现

可靠性、灵活的路由、集群、事务、高可用的队列、消息排序、问题追踪、可视化管理工具、插件系统等等。

RabbitMQ / Kafka 最好，ActiveMQ 次之，ZeroMQ 最差。当然ZeroMQ 也可以做到，不过自己必须手动写代码实现，代码量不小。尤其是可靠性中的：持久性、投递确认、发布者证实和高可用性。

1.4高并发

毋庸置疑，RabbitMQ 最高，原因是它的实现语言是天生具备高并发高可用的erlang 语言，时效性us级，延时很低。

1.5比较关注的比较

RabbitMQ 比Kafka 成熟，在可用性上，稳定性上，可靠性上，RabbitMQ 胜于  Kafka（理论上）。另外，Kafka 的定位主要在日志等方面， 因为Kafka 设计的初衷就是处理日志的，可以看做是一个日志（消息）系统一个重要组件，针对性很强，所以如果业务方面还是建议选择 RabbitMQ 。Kafka 的性能（吞吐量、TPS ）比RabbitMQ 要高出来很多。

3、使用场景

场景介绍

比如在双11的时候，当凌晨大量的秒杀和抢购商品，然后去结算的时候，就会发现，界面会提醒我们稍等，以及一些友好的图片文字提醒。而不是像前几年的时代，动不动就页面卡死，报错等来呈现给用户。

在这业务场景中，我们就可以采用队列的机制来处理，因为同时结算就只能达到这么多。

在我们平时的超市中购物也是一样，当我们在结算的时候，并不会一窝蜂一样涌入收银台，而是排队结算。这也是队列机制。一个接着一个的处理，不能插队。

1.1异步处理

比如用户注册后，需要发送注册邮件或者短信，可以将此发送过程通过MQ处理，提高并行处理的时间。

1.2应用解耦

比如电商业务中，用户下单后，需要更新库存，系统单系统和库存系统是独立的业务系统，传统的做法，直接调用库存系统接口来更新库存，有以下缺点：

库存系统无法访问时，直接影响用户下单；

库存系统与订单系统耦合性太强。

加入消息队列后，用户下单后可以不用关心其他的后续操作，实现不同业务系统之间的应用解耦。

1.3流量削峰

一般应用于秒杀或者团购抢购活动中，秒杀活动一般会因为流量过高，导致影响系统无法响应，加入消息队列能够控制活动的人数，缓解短时间内高流量压垮应用服务。

1.4日志处理

主要用于业务数据的日志采集和接收，比如在统计pv、uv等数据时，直接做收集日志的埋点。

1.5消息通讯

消息队列一般内置了消息通信机制，因此可以用作消息通讯，必须点对点通讯、聊天室通讯（发布订阅模式）两种消息模式。

4、demo演示

1） Nuget安装 RabbitMQ.Client，此次使用版本5.1.0

2）生产端

 public class MQClientController : Controller
    {
        private string queueName1 = "queue.test1";
        private string queueName2 = "queue.test2";
        private string exchangeName_direct = "exchange.direct.test1";//direct
        private string exchangeName_topic = "exchange.topic.test1";//topic
        private string exchangeName_fanout = "exchange.fanout.test1";
        private string routingKey = "queue.test.info";
        public IActionResult Index(string message = "默认")
        {
            SendMsg(message);
            return Content("发送成功");
        }
        public IActionResult Index2()
        {
            for (int i = 0; i < 100; i++)
            {
                SendMsg($"消息内容" + i);
            }
            return Content("发送成功");
        }

        private void SendMsg(string message)
        {
            var conFactory = new ConnectionFactory//创建连接工厂对象
            {
                HostName = "192.168.1.10",//IP地址
                Port = 5672,//端口号
                UserName = "administrator",//用户账号
                Password = "123",//用户密码
                AutomaticRecoveryEnabled = true,//要启用自动连接恢复
                RequestedHeartbeat = 60,//设置心跳超时时间
                NetworkRecoveryInterval = TimeSpan.FromSeconds(10),//如果恢复由于异常（例如RabbitMQ节点仍然不可达）失败，将在固定的时间间隔（默认为5秒）后重试
            };
            using (var con = conFactory.CreateConnection())//创建连接对象
            {
                using (var channel = con.CreateModel())//创建连接会话对象
                {
                    //设置类型 Direct
                    channel.ExchangeDeclare(exchangeName_direct, ExchangeType.Direct);

                    //设置Queue，此处可以不用每次发送是声明设置
                    channel.QueueDeclare(
                      queue: queueName1,//消息队列名称
                      durable: true,//是否队列持久化,需要保持和服务器上设置一致，设置为true在服务器重启时，能够存活
                      exclusive: false,
                      autoDelete: false,
                      arguments: null//参数使用说明：https://www.jianshu.com/p/2d696857ecac
                       );
                    channel.QueueDeclare(queue: queueName2, durable: true, exclusive: false, autoDelete: false, arguments: null);
                    //绑定exchage和queue
                    //fanout “无路由交换机”,直接绑定到队列的
                    //channel.QueueBind(queueName1, exchangeName_fanout, "");
                    //channel.QueueBind(queueName2, exchangeName_fanout, "");

                    //direct
                    channel.QueueBind(queueName1, exchangeName_direct, routingKey);
                    //channel.QueueBind(queueName2, exchangeName_direct, routingKey);

                    //topic
                    //channel.QueueBind(queueName1, exchangeName_topic, "queue.*.info");//queue.*,*.test
                    //channel.QueueBind(queueName2, exchangeName_topic, "queue.*.err");
                    //if (message.Contains("err"))
                    //{
                    //    //进入消费者2
                    //    routingKey = "queue.test2.err";
                    //}

                    //启用确认选择模式
                    channel.ConfirmSelect();

                    //消息内容
                    var msg = "{"key":"" + DateTime.Now.ToString("yyyyMMddHHmmssfff") + "","msg":"" + message + ""}";
                    byte[] body = Encoding.UTF8.GetBytes(msg);

                    //发送消息
                    var props = channel.CreateBasicProperties();
                    //props.Persistent = true;或者直接设置持久化
                    props.DeliveryMode = 2;//消息持久化 1不持久化 2开启持久化
                    channel.BasicPublish(exchange: exchangeName_direct, routingKey: routingKey, basicProperties: props, body: body);

                    //当消息被确认已发送，删除缓存数据
                    if (!channel.WaitForConfirms())
                    {
                        //记录发送失败的消息 ，后台重新入列处理
                    }
                }
            }
        }
    }

3）消费端

class Program
    {
        public static readonly string queueName = "queue.test1";
        static void Main(string[] args)
        {
            Console.WriteLine("消费者1-Start");
            var connFactory = new ConnectionFactory//创建连接工厂对象
            {
                HostName = "192.168.1.10",//IP地址
                Port = 5672,//端口号
                UserName = "administrator",//用户账号
                Password = "",//用户密码
                AutomaticRecoveryEnabled = true,//要启用自动连接恢复
                RequestedHeartbeat = 60,//设置心跳超时时间
                NetworkRecoveryInterval = TimeSpan.FromSeconds(10),//如果恢复由于异常（例如RabbitMQ节点仍然不可达）失败，将在固定的时间间隔（默认为5秒）后重试
            };
            //连接mq
            var conn = connFactory.CreateConnection();
            var channel = conn.CreateModel();

            //设置Queue，如果还未创建时，直接监听会报错（404找不到队列）
            channel.QueueDeclare(
              queue: queueName,//消息队列名称
              durable: true,//是否队列持久化,需要保持和服务器上设置一致，设置为true在服务器重启时，能够存活
              exclusive: false,
              autoDelete: false,
              arguments: null//参数使用说明：https://www.jianshu.com/p/2d696857ecac
               );

            //设置perfetchCount=1 。这样就告诉RabbitMQ 不要在同一时间给一个工作者发送多于1个的消息
            //设置prefetchCount: 1来告知RabbitMQ，在未收到消费端的消息确认时，不再分发消息，也就确保了当消费端处于忙碌状态时，不再分配任务。
            channel.BasicQos(0, 1, false);

            //创建消费者对象
            var consumer = new EventingBasicConsumer(channel);

            //监听消费事件
            consumer.Received += (model, ea) =>
            {
                byte[] message = ea.Body;//接收到的消息
                Console.WriteLine("消费者1-接收到信息为:" + Encoding.UTF8.GetString(message));

                //通过数据库操作后通知MQ删除消息
                channel.BasicAck(ea.DeliveryTag, false);//手动应答MQ已经成功接收

                //channel.BasicNack(ea.DeliveryTag, false, true);//支持拒绝接收多条消息，是否重回队列
                //channel.BasicReject(ea.DeliveryTag, true);//拒绝接收单条消息，是否重回队列

                //队列堵塞测试
                //System.Threading.Thread.Sleep(10000);
            };
            //监听shutdown事件
            consumer.Shutdown += (model, e) =>
            {
                //记录日志提醒
                Console.WriteLine("消费者1-Shutdown被执行");
            };
            //消费者开启监听
            channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
            Console.ReadKey();
        }
    }

cunsumer端可以部署多个服务来消费。

5、总结

5.1 使用消息队列可以降低应用系统间的耦合度，但增加了系统的复杂性，可用性也随之降低（此处是指复杂度造成的可用性）。

5.2 异步消息能够充分提升系统性能（吞吐量、TPS），提高程序并发性能。

 以上内容部分说明内容摘自网上。