RabbitMQ学习笔记2-理解消息通信

Posted 2020-08-14 自行车上的程序员

消息包含两部分：1、有效载荷(payload) - 你想要传输的数据。2、标签(lable) - 描述有效载荷的相关信息，包含具体的交换器、消息的接受兴趣方等。

rabbitmq的基础流程如下：

RabbitMQ的客户端和服务端通过channel与RabbitMQ服务器进行通信。

Channel(信道)：程序和RabbitMQ之间的连接是通过channel，channel是基于TCP协议之上的​，
一个TCP连接可以有多个channel，可以比喻成如下：一根大电缆里面有很多根小的电线，大电缆就是一个TCP连接，
而小电线就是channel了，而为什么要这么设计呢？因为系统创建和销毁TCP连接是很重量级的操作，
基于一个TCP上进行多个不冲突的通信操作可以大大增加效率。

 

AMQP的消息路由必须包含3个部分：交换器(Exchange)、队列（Queue）、绑定(Bind)。

1、Queue(队列)：用来存放消息的地方，等待消费者来获取。消费者可以通过以下2种方式从queue里接收消息。

⑴、通过AMQP的basic.consume命令订阅。这样会将channel置为接收模式，直到取消对该queue的订阅为止。订阅了消息后，消费者在消费(或者拒绝)最近接收的那条消息后，就能从队列中（有可用的）自动接收下一条信息。此订阅模式生产者把消息发布到queue后就会被消费消费者自动接收到。

⑵、通过AMQP的basic.get命令获取下一条消息。该模式一次只获取一条信息，然后就取消了订阅，如果需要继续获取下一条需要再次发送此命令。注意：不应该在for循环里使用basic.get来替代basic.consume，这样会影响Rabbit的性能。

 当多个消费者订阅同一个队列时，rabbit会采取循环的方式发送给消费者，每条消息只发给一个订阅的消费者，消费者收到信息后需要发送basic.ack来确认收到，或者将auto_ack设置为true来自动确认。如果消费者未回复确认信息，rabbit将不会再对此消费者发送消息了。当然你可以使用basic.reject命令来拒绝rabbit发来的消息，拒绝的同时将该消息从队列里删除。

流程如下：

说了这么多，队列该如何创建呢？使用queue.declare命令来创建队列，需要注意的是在同一条Channel(信道)上只能订阅一个队列，如果在Channel上已经订阅了一个队列的情况需要订阅另外一个队列，唯一的处理方法就是取消订阅之前的队列并将Channel置为【传输模式】然后再订阅新的。

该命令主要用到的参数有：

/**
     * Declare a queue
     * @param queue 队列的名称,如果为空将生成随机的名称
     * @param durable 是否持久化的队列 (服务重启后队列和消息都还在)
     * @param exclusive 是否是私有(独占)的队列 (限制在此连接)
     * @param autoDelete 是否当没人使用队列时自动删除
     * @param arguments other properties (construction arguments) for the queue
     * @return a declaration-confirm method to indicate the queue was successfully declared
     * @throws java.io.IOException if an error is encountered
     */
    Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
                                 Map<String, Object> arguments) throws IOException;

 

Exchange(交换器)与绑定：服务器根据路由键(规则)将消息路由到指定的队列。Exchange与queue通过绑定操作来进行关联。

交换器共有4种类型：direct、fanout、topic和headers。

 direct exchange：如果路由键匹配的话，消息会被投递到对应队列。服务器必须实现direct类型的交换器，包含一个空白字符串名称的默认交换器。

如下的代码将使用到默认对应exchange类型的exchange

channel.basicPublish("",ROUTING_KEY, null, msg.getBytes());

代码中使用了默认的direct交换器，路由键为队列的名称。

当默认的direct交换器无法满足需求时可以使用exchange.declare命令设置合适的参数来创建一个direct交换器。

别人对此交换器的解释：

Direct Exchange – 处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。
如果一个队列绑定到该交换机上要求路由键 “dog”，则只有被标记为“dog”的消息才被转发，不会转发dog.puppy，也不会转发dog.guard，只会转发dog。

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

topic exchange：通过模式匹配分析消息的routing-key(路由键)属性。它将routing-key和binding-key的字符串切分成单词。这些单词之间用点隔开。它同样也会识别两个通配符：#匹配0个或者多个单词，*匹配一个单词。例如，binding key：*.stock.#匹配routing key：usd.stcok和eur.stock.db，但是不匹配stock.nana

 

注意：如果将消息发送给未绑定/不符合路由规则的Queue，则消息会进入“黑洞”(Rabbit的一个特殊的队列，老版本的就是直接丢失了)里。

 

RabbitMQ系统自带的几个Exchange：