RabbitMQ巩固学习一

Posted 2020-12-11 zhangnever

说起RabbitMQ大家第一时间应该想到的就是异步队列，关于异步队列的话题简直太多了，各位同学在园子里一搜便知。我第一次听异步队列这个名词感觉非常高大上??,想到这项技术必须要学。但是学习的任何一门技术没经过项目的洗礼,都似乎少了点什么。嗯。是的。只有在企业级开发中,才能找到自己的遗漏的知识点。所以大家一定要想办法将自己学习到的东西糅合进项目中。以我现在一贯的作风就是学习、巩固、总结与分享。虽然介绍RabbitMQ这类博客已经很多了,但是人家写的毕竟是人家写的,我要把自己的学习心得记录下来,一来巩固自己的知识,二来可以帮到其它的小伙伴。??   

首先说一下队列.队列的本质是一种先进先出的线性存储结构。注意不要和栈弄反了,栈的存储结构是先进后出

 那怎样理解异步队列？

举个例子：假如在电商系统中,用户下单成功之后。A服务要拿到下单信息发送邮件(350ms)给用户,B服务要拿到下单信息发送短信(400ms)给用户,C服务要拿到下单信息记录日志(300ms)，如果三个服务利用RPC远程调用下单服务提供的api,在排队等候时(可能有分布式锁)将要耗费1050ms,而且耦合太高了,如果下单服务改由另一个服务承担那么ABC服务都得跟着改,接下来就变成下面这样。

 用户下单成功后直接将数据存进消息队列(数据库也会存一份)，ABC三服务直接从消息队列中拿下单信息，不再排队等候(异步),时间耗费400ms。性能提升了一倍之多。而且谁将数据放入消息队列ABC三服务根本不用管,它们现在只认数据不认人.??

再来说一下RabbitMQ.RabbitMQ是采用Erlang语言实现的消息中间件。RabbitMQ几乎支持所有的常用语言，并且提供了用户管理界面,可以方便用户管理和监控消息。在使用RabbitMQ之前我们先安装她。这里的话我只说一下在Linux下使用小鲸鱼安装

docker pull rabbitmq:management
docker run -d -p 15672:15672 -p 5672:5672 --name rabbitmq rabbitmq:management

不出意外的话容器已经启动了,如果有问题的话就看日志.

 默认情况下访问RabbitMQ 用户名和密码都是 "guest" ,但是这个这个账户有限制，默认只能通过本地网络访问,远程连接的话受限.

这里的话要执行以下命令.  ps:命令是网上找的??

docker exec -it rabbitmq bash --进入容器
rabbitmq-plugins enable rabbitmq_management --开启插件命令

这里再说一下生产者和消费者。生产者就是往消息队列中加入(创建)数据的一方。消费者就是拿到(接收)消息的一方。

接下来在代码中简单演示一下.

新建两个控制台程序。引入包RabbitMQ.Client.

生产者代码：

static void Main(string[] args)
{
    //创建连接工厂
    var factory = new ConnectionFactory
    {
        UserName = "guest",//用户名
        Password = "guest",//密码
        HostName = "***.**.***.***",//rabbitmq ip
        Port = 5672,//端口号
    };
    //创建连接
    var connection = factory.CreateConnection();
    //创建信道
    var channel = connection.CreateModel();
    //创建一个队列,名称为:RabbitMQStudy
    channel.QueueDeclare("RabbitMQStudy", false, false, false, null);
    Console.WriteLine("已连接到RabbitMQ.^-^,可向队列投递消息!");
    Console.WriteLine("输入close则关闭连接");
    string text = string.Empty;
    bool flag = true;
    while (flag)
    {
        Console.WriteLine("请输入数据......");
        text = Console.ReadLine();
        switch (text)
        {
            case "close":
                flag = false;
                continue;
            default:
                flag = true;
                break;
        }
        var bytes = Encoding.UTF8.GetBytes(text);
        //发布消息
        channel.BasicPublish("", "RabbitMQStudy", null, bytes);
        Console.WriteLine("消息已发送成功!");
        Console.WriteLine(new string(‘-‘, 100));
    }
    Console.WriteLine("连接已关闭......");
    //关闭资源
    channel.Close();
    //释放连接
    connection.Close();
}

消费者代码:

static void Main(string[] args)
{
    //创建连接工厂
    ConnectionFactory factory = new ConnectionFactory
    {
        UserName = "guest",//用户名
        Password = "guest",//密码
        HostName = "***.**.***.***",//rabbitmq ip
        Port = 5672,//端口号
    };
    //创建连接
    var connection = factory.CreateConnection();
    //创建信道
    var channel = connection.CreateModel();

    //var consumer = new DefaultBasicConsumer(channel);
    var consumer = new EventingBasicConsumer(channel);
    //为消费者送达消息时产生的事件
    consumer.Received += (ch, ea) =>
    {
        var message = ea.Body.ToArray();//接收到的消息
        Console.WriteLine($"接收到信息:{Encoding.UTF8.GetString(message)}");
        //消息已被消费
        channel.BasicAck(ea.DeliveryTag, false);
        Console.WriteLine("消息已成功接收并消费!");
        Console.WriteLine(new string(‘-‘, 100));
    };
    //启动消费者并设置为手动应答
    String consumerTag = channel.BasicConsume("RabbitMQStudy", false, consumer);
    Console.WriteLine("消费者已启动成功!");
    Console.ReadKey();
    channel.Dispose();
    connection.Close();
} 

开启了三个消费者,这时队列中的消息会被平均分摊(订阅了同一个队列).RabbitMQ有轮询的机制，所以不是每个消费者都可以收到所有的消息井处理。

下面说一下三个重要的知识点:交换器、路由键、绑定

交换器:生产者向消息队列发送消息数据时,其实消息先到交换器再由交换器路由到相应的队列。这里有同学就会问了,我们刚刚上面的代码并没有申明交换器啊,其实这里我们发送消息时exchange参数为(" ")，相当于默认创建好的一个交换器(类型为direct)。直接用RouteKey指定队列名即可.

交换器一共有四种类型:1.fanout:此类型的交换器会将发送到该交换器的的消息路由到所有与该交换器绑定的队列中
                  2.directc:如果交换器类型为direct类型,那么RoutingKey(路由键,相当于消息投送至与交换器绑定的哪个队列)和BindingKey(绑定键,交换器与队列相绑定有一个BindingKey)需要完全匹配(相同)。
                  3.topic:topic是升级版的direct类型的交换器,与direct类型交换器相同的是topic类型的交换器也是将消息路由到RoutingKey与BindingKey相匹配的队列中。与direct类型交换器不同的是topic类型的交换器有模糊匹配。 例如:我们发送消息时会路由到指定RouteKey队列中.交换器在与队列绑定的时候,BindingKey可以是模糊类型的.RouteKey和BindingKey都是以点号“.”分割的字符串.BindingKey的字符串可以存在“ * ”和“ # ”特殊字符。在以点号“ . ”为分割的字符串中，“ . ”用于匹配一个单词," # "用于匹配多个(或零个)单词.
                  4.headers:这个交换器我在这里不做介绍,我没用过??,上面介绍的三个是最常用的了。

如下图所示申明了一个topic类型的交换器,当路由键为core.never.net的消息会同时路由到队列1和队列2中。当路由键为go.never.cn的消息也会同时路由到队列1与队列2中。当路由键为www.bilibili.com的消息会路由到队列2中。

 

 举个例子,在生产者代码中再加几句代码：

//和上面一样,代码省略...
//创建一个队列,名称为:RabbitMQStudy
channel.QueueDeclare("RabbitMQStudy", false, false, false, null);
//创建一个队列,名称为:RabbitMQStudy1
channel.QueueDeclare("RabbitMQStudy1", false, false, false, null);
//声明一个交换器
channel.ExchangeDeclare("exange", "topic", true, false, null);
//绑定
channel.QueueBind("RabbitMQStudy", "exange","#.routeKey");
//绑定
channel.QueueBind("RabbitMQStudy1", "exange", "*.com");

//和上面一样,代码省略...

//发布消息 
channel.BasicPublish("exange", "core.routeKey", null, bytes); 

//和上面一样,代码省略...

在建一个控制台程序(消费者),和第一个消费者的代码一样,只不过两个消费者监听设置的队列名不一样。一个是RabbitMQStudy，一个是RabbitMQStudy1。

 //启动消费者并设置为手动应答
 String consumerTag = channel.BasicConsume("RabbitMQStudy1", false, consumer);

运行... ...

可以看到只有消费者1接收到了消息,模糊匹配的作用就展现出来了.

 路由键：生产者将消息发送给交换器时,会指定一个路由key，用于设定这个消息的路由规则，路由key需要与交换器类型和绑定key联合使用才能有效。

绑定：RabbitMQ中通过绑定将交换器与队列进行关联,在交换器与队列进行绑定时会有一个绑定键,这样结合路由键RabbitMQ就会知道将消息投递到哪个队列中。

争对上述代码及文字我们来总结一下

消息队列的优缺点：1.解耦,我上面提到了,消费者只需要拿数据,数据的投递方是谁它一点都不过问。
               2.削峰.当我们设计一个秒杀业务的时候,加入成千上万的请求涌入,服务器的压力过大。为了缓解压力我们可以将请求先放入队列中,服务器一定时间内可以承受多少请求就拿多少。
               3.异步.我上面也提到了.直接从消息队列中拿数据,无需排队等候,大大提高了系统性能。

RabbitMQ的使用过程:生产者：1.先创建连接工厂,与RabbitMQ建立连接。
                        2.创建信道。
                        3.声明一个交换器,设置交换器类型、是否持久化等。
                        4.声明队列,设置队列是否排他、是否持久化等。
                        5.将交换器与队列绑定,设置RouteKey。
                        6.发送消息至交换器,并设置交换器名称、路由键、消息主体等。
                        7.交换器根据路由键和绑定的队列相匹配,投送消息至队列中。
                        8.如果没有找到相应队列,会根据生产者配置的属性丢弃消息或者退回给生产者。
                        9.关闭连接。

                  消费者:1.先创建连接工厂,与RabbitMQ建立连接。
                        2.创建信道。
                        3.等待生产者投递消息至队列,消费者接收消息。
                        4.消费者确认接收到消息
                        5.RabbitMQ从队列中删除已确认的消息。
                        6.关闭连接。

今天就这么多,后续文章会一一介绍RabbitMQ其它特性。

如有不足,请见谅！