RabbitMQ学习笔记（一）基本概念知识

小灰灰 2021-01-24 18:43 820阅读 0赞

# 整体架构 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70][]

# Queue #

Queue（队列）在RabbitMQ的作用是存储消息，队列的特性是先进先出，由上图可知，通过 clientA、B(生产者)发送的消息经过RabbitMQ Server最终被传送到Queue1、2中,而client1、2、3(消费者)，则从队列中取出这些消息，PrefetchCount=1,是每次取出一条消息，处理完再取，防止消息堆积，通俗的讲就是消费者有多大本事，就干多少活，消费者处理的越慢，其消息分配分发的就少，反之消费者消息处理的多，处理的快，就可以多向这个消费者分配一些消息。服务端给客户端发消息的时候，先检查一下，这个消费者现在还有多少消息，如果处理的消息超过1条，就不给这个消费者发送消息了

队列消息持久化也是一个重要的概念，默认是开启的

1.  RabbitMQ在服务端没有声明队列和消息持久化时，队列和消息是存在内存中的，服务端宕机了，队列和消息也不会保留。
2.  服务端声明持久化，客户端想接受消息的话，必须也要声明queue时，也要声明持久化，不然的话，客户端执行会报错。

# Exchange #

Exchange(交换机，或路由器)，作用和名字一样，就是负责把消息路由到相应的队列中，这里其实是发布/订阅的设计模式思想，消费者告诉exchange

我需要什么类型的信息，当生产者消息到来的时候，exchange就把消息通知到对应的队列中。当然，exchange也有自己的路由规则(匹配规则),即

# exchangeType #

|          |                                                              |
| -------- | ------------------------------------------------------------ |
| 类型名称 | 类型描述                                                     |
| fanout   | 把所有发送到该Exchange的消息路由到所有与它绑定的Queue中      |
| direct   | Routing Key==Binding Key                                     |
| topic    | 模糊匹配                                                     |
| headers  | Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。 |

#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 key #

在绑定（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。

#RPC#

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

RabbitMQ中实现RPC的机制是：

*  客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14种properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
 *  服务器端收到消息并处理
 *  服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性
 *  客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理

#应用场景#

**应用场景****1**

一个P向queue发送一个message，一个C从该queue接收message并打印。![20181123092252350.png][]

**应用场景****2-work queues**

将耗时的消息处理通过队列分配给多个consumer来处理，我们称此处的consumer为worker，我们将此处的queue称为Task Queue，其目的是为了避免资源密集型的task的同步处理，也即立即处理task并等待完成。相反，调度task使其稍后被处理。也即把task封装进message并发送到task queue，worker进程在后台运行，从task queue取出task并执行job，若运行了多个worker，则task可在多个worker间分配。

![20181123092252347.png][]

**应用场景****3-Publish/Subscribe**

在应用场景2中一个message(task)仅被传递给了一个comsumer(worker)。现在我们设法将一个message传递给多个consumer。这种模式被称为publish/subscribe。此处以一个简单的日志系统为例进行说明。该系统包含一个log发送程序和一个log接收并打印的程序。由log发送者发送到queue的消息可以被所有运行的log接收者接收。因此，我们可以运行一个log接收者直接在屏幕上显示log，同时运行另一个log接收者将log写入磁盘文件。

**应用场景****4-Routing**

应用场景3中构建了简单的log系统，可以将log message广播至多个receiver。现在我们将考虑只把指定的message类型发送给其subscriber，比如，只把error message写到log file而将所有log message显示在控制台。

**应用场景****5-topic**

应用场景4中改进的log系统中用direct类型的exchange替换应用场景3中的fanout类型exchange实现将不同的log message发送给不同的subscriber（也即分别通过不同的routing\_key将queue绑定到exchange，这样exchange便可将不同的message根据message内容路由至不同的queue）。但仍然存在限制，不能根据多个规则路由消息，比如接收者要么只能收error类型的log message要么只能收info类型的message。如果我们不仅想根据log的重要级别如info、warning、error等来进行log message路由还想同时根据log message的来源如auth、cron、kern来进行路由。为了达到此目的，需要topic类型的exchange。topic类型的exchange中routing\_key中可以包含两个特殊字符：“\*”用于替代一个词，“\#”用于0个或多个词。![20181123092352135.png][]

应用场景6-PRC

在应用场景2中描述了如何使用work queue将耗时的task分配到不同的worker中。但是，如果我们task是想在远程的计算机上运行一个函数并等待返回结果呢。这根场景2中的描述是一个完全不同的故事。这一模式被称为远程过程调用。现在，我们将构建一个RPC系统，包含一个client和可扩展的RPC server，通过返回斐波那契数来模拟RPC service。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70 1][]

#问题#

**1.将queue，exchange, message等都设置了持久化之后就能保证100%保证数据不丢失了嚒？**

答案是否定的。

首先，从consumer端来说，如果这时autoAck=true，那么当consumer接收到相关消息之后，还没来得及处理就crash掉了，那么这样也算数据丢失，这种情况也好处理，只需将autoAck设置为false(方法定义如下)，然后在正确处理完消息之后进行手动ack（channel.basicAck）.

String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

其次，关键的问题是消息在正确存入RabbitMQ之后，还需要有一段时间（这个时间很短，但不可忽视）才能存入磁盘之中，RabbitMQ并不是为每条消息都做fsync的处理，可能仅仅保存到cache中而不是物理磁盘上，在这段时间内RabbitMQ broker发生crash, 消息保存到cache但是还没来得及落盘，那么这些消息将会丢失。那么这个怎么解决呢？首先可以引入RabbitMQ的mirrored-queue即镜像队列，这个相当于配置了副本，当master在此特殊时间内crash掉，可以自动切换到slave，这样有效的保障了HA, 除非整个集群都挂掉，这样也不能完全的100%保障RabbitMQ不丢消息，但比没有mirrored-queue的要好很多，很多现实生产环境下都是配置了mirrored-queue的。还有要在producer引入事务机制或者Confirm机制来确保消息已经正确的发送至broker端， RabbitMQ的可靠性涉及producer端的确认机制、broker端的镜像队列的配置以及consumer端的确认机制，要想确保消息的可靠性越高，那么性能也会随之而降，鱼和熊掌不可兼得，关键在于选择和取舍。

**2.消息什么时候刷到磁盘？**

写入文件前会有一个Buffer,大小为1M,数据在写入文件时，首先会写入到这个Buffer，如果Buffer已满，则会将Buffer写入到文件（未必刷到磁盘）。

有个固定的刷盘时间：25ms,也就是不管Buffer满不满，每个25ms，Buffer里的数据及未刷新到磁盘的文件内容必定会刷到磁盘。

每次消息写入后，如果没有后续写入请求，则会直接将已写入的消息刷到磁盘：使用Erlang的receive x after 0实现，只要进程的信箱里没有消息，则产生一个timeout消息，而timeout会触发刷盘操作。

**3.如果在消息到达broker之前已经丢失的话，持久化操作也解决不了这个问题，因为消息根本就没到达代理服务器，你怎么进行持久化，那么这个问题该怎么解决呢？**

RabbitMQ为我们提供了两种方式：

1.  通过AMQP事务机制实现，这也是AMQP协议层面提供的解决方案；
2.  通过将channel设置成confirm模式来实现；

**事务机制**

RabbitMQ中与事务机制有关的方法有三个：txSelect(), txCommit()以及txRollback(), txSelect用于将当前channel设置成transaction模式，txCommit用于提交事务，txRollback用于回滚事务，在通过txSelect开启事务之后，我们便可以发布消息给broker代理服务器了，如果txCommit提交成功了，则消息一定到达了broker了，如果在txCommit执行之前broker异常崩溃或者由于其他原因抛出异常，这个时候我们便可以捕获异常通过txRollback回滚事务了。

**Confirm****模式**

生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者（包含消息的唯一ID）,这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会将消息写入磁盘之后发出，broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理。

confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息。

在channel 被设置成 confirm 模式之后，所有被 publish 的后续消息都将被 confirm（即 ack） 或者被nack一次。但是没有对消息被 confirm 的快慢做任何保证，并且同一条消息不会既被 confirm又被nack 。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70]: /images/1611514095020.png
[20181123092252350.png]: /images/1611514075205.png
[20181123092252347.png]: /images/1611514055860.png
[20181123092352135.png]: /images/1611514036007.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70 1]: /images/1611514015956.png