RabbitMq 学习二

消息中间件：

• 消息：软件之间传递通讯时需要传送的数据
• 中间件：本来软件A和软件B之间可以直接传递消息，但是，非要先让软件A把消息传给中间件，然后再有中间件把消息传给软件B
• 消息队列：是消息中间件的一直实现方式

消息中间件：可以理解为软件与软件之间交互的消息在消息中间件中进行存储和管理的一种容器

消息队列：是消息中间件的一种实现方式

  常见的消息队列中间件: ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ等。

 

使用消息中间件的优点：

•  异步处理，提高响应速度

一个客户端请求发送进来，系统A会调用系统B、C、D三个系统，同步请求的话，响应时间就是系统A、B、C、D的总和，也就是800ms。如果使用MQ，系统A发送数据到MQ，然后就可以返回响应给客户端，不需要再等待系统B、C、D的响应，可以大大地提高性能。对于一些非必要的业务，比如发送短信，发送邮件等等，就可以采用MQ。比如我们最常见的短信验证码功能,当我们在界面点击“获取验证码”后,我们还可以同时进行其他的操作,如输入更新的密码等,此时,我们不需要一直等到手机收到短信了才进行下一步的操作,这就是异步处理,提高了用户体验。

• 应用解耦:

如图所示。假设有系统B、C、D都需要系统A的数据，于是系统A调用三个方法发送数据到B、C、D。这时，系统D不需要了，那就需要在系统A把相关的代码删掉。假设这时有个新的系统E需要数据，这时系统A又要增加调用系统E的代码。为了降低这种强耦合，就可以使用MQ，系统A只需要把数据发送到MQ，其他系统如果需要数据，则从MQ中获取即可。

• 流量削峰:

如常见的秒杀系统,这其实是MQ一个很重要的应用。假设系统A在某一段时间请求数暴增，有5000个请求发送过来，系统A这时就会发送5000条SQL进入MySQL进行执行，MySQL对于如此庞大的请求当然处理不过来，MySQL就会崩溃，导致系统瘫痪。如果使用MQ，系统A不再是直接发送SQL到数据库，而是把数据发送到MQ，MQ短时间积压数据是可以接受的，然后由消费者每次拉取2000条进行处理，防止在请求峰值时期大量的请求直接发送到MySQL导致系统崩溃。

 为什么使用RabbitMq

 1：RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。

AMQP是跨平台的，跨语言的，JMS不是，只能用于Java平台，考虑到现在流行的微服务开发,不同微服务可能使用的是不同的语言开发,综合考虑选择RabbitMQ消息队列,因为它是AMQP规范的一个实现,可以达到跨平台、跨语言。

RabbitMq的服务器页面：

• 5672: 默认的客户端连接的端口
• 15672: 默认的web管理界面的端口
• 默认账号密码都为: guest

 RabbitMq的核心概念：

• Exchange：交换器，用来接受生产者发送的消息并将这些消息路由给服务器中的队列 有4种类型： direct(默认)， fanout, topic, headers，不同类型的Exchange转发消息的 策略有所区别。
• Queue：消息队列，用来存放消息直到发送给消费者。他是消息队列的容器，也是消息的终点。一个消息可以发给一个或者多个消息队列，消息一直都在队列里面，直到消费者连接到这个队列把他取走。
• Channel：信道，多路复用连接中的一条独立的双向数据流通道，信道是建立在真实的tcp连接内的虚拟连接，AMQP命令都是通过信道发送出去的，不管是发布消息、订阅队列还是接受消息，这些动作都是通过信道完成的，因为对于操作系统来说，建立和销毁TCP都是非常昂贵的开销，所以引入信道的概念，以复用一条TCP连接
• Publish：消息的生产者，也是一个向交换机发步消息的客户端应用程序
• Consumer:消息的消费者，表示一个从消息队列中取得消息的客户端应用程序
• Virtual Host：虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。 vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 
• Binding(绑定): 用于消息队列和交换器之间的关联,一个绑定就是基于路由键将交换机和消息队列连接起来的路由规则,交换器跟队列的绑定可以是多对多的关系

创建一个发送消息客户端：

static void Main(string[] args)
        {
            //定义连接工厂
            var factory = new ConnectionFactory();
            factory.HostName = "localhost";//RabbitMQ服务在本地运行
            factory.UserName = "guest";//用户名
            factory.Password = "guest";//密码
       
            //创建连接
            using (var connection = factory.CreateConnection())
            {
                //建立信道
                using (var channel = connection.CreateModel())
                {
                    //声明队列
                    channel.QueueDeclare("mymessage", false, false, false, null);//创建一个名称为hello的消息队列
                    Console.WriteLine("输入需要传输的消息，输入Exit退出");
                    var message = Console.ReadLine();
                    while (message != "Exit")
                    {
                        var body = Encoding.UTF8.GetBytes(message);
                        //发送消息
                        channel.BasicPublish(exchange: "",
                                             routingKey: "mymessage",
                                             basicProperties: null,
                                             body: body);
                        Console.WriteLine(" 发送消息 {0}", message);
                        message = Console.ReadLine();
                    }
                  
                }
            }
            Console.WriteLine("按回车退出");
            Console.ReadLine();
        }

创建一个接受消息客户端：

static void Main(string[] args)
        {
            //定义连接工厂
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//RabbitMQ服务在本地运行
                UserName = "guest",//用户名
                Password = "guest",//密码
                
            };
            //创建连接
            using (var connection = factory.CreateConnection())
            {
                //创建信道
                using (var channel = connection.CreateModel())
                {
                    //声明通道
                    channel.QueueDeclare(queue: "mymessage",
                                                  durable: false,
                                                  exclusive: false,
                                                  autoDelete: false,
                                                  arguments: null);
                    //定义消费者
                    var consumer = new EventingBasicConsumer(channel);
                    //接收消息
                    consumer.Received += (model, ea) =>
                    {
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body.ToArray());
                        Console.WriteLine("收到消息 {0}", message);
                    };
                    //监听队列
                    channel.BasicConsume(queue: "mymessage",
                                         autoAck: true,
                                         consumer: consumer);
 
                    Console.WriteLine(" 按回车退出");
                    Console.ReadLine();
                }
 
            }
 
        }

这是最简单的消息模型，一个生产者，一个消费者

 订阅模式

• 1一个生产者，多个消费者
• 每个消费者都有自己的一个队列
• 生产没有将消息直接发送到队列，而是发送到了交换机
• 每个队列都要绑定交换机
• 生产者发送的消息，经过到达交换机到达队列，实现一个消息被多个消费者获取的目的

交换机（Exchange）的模式有以下几种

Fanout：广播，交换机将消息发送到所有与之绑定的队列中去

Direct：定向，交换机按照指定的Routing Key发送到匹配的队列中去

Topics:通配符，与Direct大致相同，不同在于Routing key可以根据通配符进行匹配

• 订阅模型之Direct

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。 fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。 fanout 类型转发消息是最快的 。

 生产者：

 //声明交换机 fanout：交换机类型 主要有fanout,direct,topics三种
 channel.exchangeDeclare(EXCHANGE_NAME,"fanout");

 消费者：

//绑定队列到交换机上
channel.queueBind(QUEUE_NAME,Send.EXCHANGE_NAME,"");

• 订阅模型之Direct

在fanout模型中，生产者发布消息，所有消费者都可以获取所有消息。在路由模式（Direct）中，可以实现不同的消息被不同的队列消费，在Direct模式下，交换机不再将消息发送给所有绑定的队列，而是根据Routing Key将消息发送到指定的队列，队列在与交换机绑定时会设定一个Routing Key，而生产者发送的消息时也需要携带一个Routing Key。

生产者：

//声明交换机 fanout：交换机类型 主要有fanout,direct,topics三种
 channel.exchangeDeclare(EXCHANGE_NAME,"direct");
//生产者发送消息时，设置消息的Routing Key:"dog"
 channel.basicPublish(EXCHANGE_NAME,"dog",null,message.getBytes());

消费者：

//绑定队列到交换机上,并制定路由键为"dog"
channel.queueBind(QUEUE_NAME, com.bw.rabbitmq.routing.Send.EXCHANGE_NAME,"dog");

• 订阅模型之Topics

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词
*：匹配不多不少恰好1个词

生产者：

 //声明交换机 topic：交换机类型 
channel.exchangeDeclare(EXCHANGE_NAME,"topic");
channel.basicPublish(EXCHANGE_NAME,"dog.1",null,message.getBytes());

 消费者1：

//绑定队列到交换机上,并制定路由键匹配规则为"dog.*"
 channel.queueBind(QUEUE_NAME, com.bw.rabbitmq.topics.Send.EXCHANGE_NAME,"dog.*");

 消费者2： 

 //绑定队列到交换机上,并制定路由键匹配规则为"#.1"
 channel.queueBind(QUEUE_NAME, com.bw.rabbitmq.topics.Send.EXCHANGE_NAME,"#.1");

消费者3：

 //绑定队列到交换机上,并制定路由键匹配规则为"cat.#"
 channel.queueBind(QUEUE_NAME, com.bw.rabbitmq.topics.Send.EXCHANGE_NAME,"cat.#");

结果：消费者1和消费者2可以收到消息，消费者3不能收到消息。