RabbitMQ研究（二）组件与协议

ゝ一纸荒年。 2022-03-26 08:24 187阅读 0赞

# 概念 #

## 生产者和消费者 ##

`Producer: 生产者，就是投递消息的一方。`

生产者创建消息，然后发布到RabbitMQ。一般分为两部分：消息体和标签。消息体被称为payload。消息体一般是一个带有业务逻辑结构的数据，比如一个JSON字符串。当然可以进一步对这个消息体进行序列化。消息的标签用来表述消息，比如一个交换器的名称和一个路由键。

`Consumer：消费者，就是接收消息的一方。`  
消费者连接到RabbitMQ服务器，并订阅到队列上。当消费者消费一条消息，只是消费消息的消息体，在路由的时候，消息的标签会丢弃，存入队列的只有消息体。

`Broker：消息中间件的服务节点。`  
一个Broker可以看作是一个服务节点，或者称为服务实例  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70]

生产者将业务方数据进行序列化，封装成消息，发送到Broker中。消费者订阅并接收消息，经过可能的解包之后得到原始数据，之后再进行业务处理。

## 队列 ##

`Queue：队列，是RabbitMQ的内部对象，用于存储消息`

RabbitMQ中的消息只存储在队列中，多个消费者可以订阅同一个队列，这时的队列消息会被平摊，给多个消费者进行处理，并不是说每个消费者都收到所有的消息处理

## 交换器、路由键、绑定 ##

`Exchange:所有的消息都发送给交换器，由交换器来选择路由。`生产者将消息发送给Exchange中，由于交换器将消息路由到一个或者多个队列中。如果路由不到，或许会返回给生产者，或许直接丢弃

`RoutingKey：路由键。`生产者将消息发送给交换器的时候，一般会指定一个RoutingKey，用来指定这个消息的路由规则，而这个RoutingKey需要与交换器类型和绑定键联合使用才能最终生效。

总结：交换器和绑定键固定的时候，生产者把消息发送到交换器中，由RoutingKey来决定消息流向哪里。

`Binding：绑定。RabbitMQ通过绑定将交换器和队列关联起来，在绑定的时候一般会指定一个绑定键（BindingKey）。`一个队列可以有多个绑定键

第一次看到这个的时候可能会有些费解，为什么要分成两个。一个就搞定了，这个就涉及到RabbitMQ的交换器类型了，`direct`确实是两个必须一致，但是如果需要匹配一批的时候，`topic模式`，后面会详细介绍

## 交换器类型 ##

RabbitMQ常用的交换器类型有`fanout`、`direct`、`topic`、`headers`四种。AMQP协议中还有两种`System`和`自定义`

*  fanout  
    不处理路由键，。会把所有的发送到该交换器的消息路由到所有与该交换器绑定的队列中。适用于广播的方式，需要将一个消息广播到所有的队列
 *  direct  
    direct类型的交换器最简单，他是将消息路由到BindingKey和RoutingKey完全匹配的队列中。就是路由键=绑定键
 *  topic  
    路由键通过类似正则的方式绑定键。direct是完全匹配BindingKey和RoutingKey，但是这种方式不能满足业务需求。可能需要以某些开头的方式来匹配等  
    匹配规则：

> 1)RoutingKey为一个点号分割独立的的单词  
> 2)BindingKey和RoutingKey也一样是点号分隔的字符串  
> 3)BindingKey中可以存在两种特殊字符串 `*`和`#`，用于模糊匹配，其中`*`用于匹配一个单词，`#`表示零个或者多个单词。如“log.\*”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.\#”能与上述两者匹配。  
> 4)如果匹配不了，通过设置mandatory=true返回给生产者，mandatory=false表示丢弃，默认为false

*  headers  
    headers类型的交换器不依赖于路由键的匹配来路由消息，而是根据发送的消息内容中的`headers`属性来匹配。在绑定队列和交换器时制定一组键值对，当发送消息到交换器的时候，RabbitMQ会获取到该消息的headers，对比其中的键值对是否完全匹配队列和交换器绑定时指定的键值对，如果完全匹配则消息会路由到该队列，否则不会路由到该队列。headers的类型性能很差，不推荐使用

## 运转流程 ##

重新梳理一下RabbitMQ的生产者发送消息的流程

1.  生产者链接到RabbitMQ Broker，建立一个连接（Connection），开启一个信道（Channel）
2.  生产者声明一个交换器，并设置相关属性，比如交换机类型、是否持久化，是否自动删除等
3.  生产者声明一个队列并设置相关属性，比如是否排他、是否持久化、是否自动删除等
4.  生产者通过路由键将交换器和队列绑定起来
5.  生产者发送消息至RabbitMQ Broker，其中包含路由键、交换器等
6.  相应的交换器根据接收到的路由键查找相匹配的队列
7.  如果找到，则从生产者发送过来的消息存入相应的队列中
8.  如果没有找到，根据mandatory属性来选择丢弃还是回退到生产者
9.  关闭信道
10. 关闭连接

消费者接收消息的流程：

1.  消费者链接到RabbitMQ Broker，建立一个连接（Connection），开启一个信道（Channel）
2.  消费者向Broker请求消费响应队列中的消息，可能会设置相应的回调函数
3.  等待Broker回应并投递相应队列中的消息，消费者接收消息
4.  消费者确认（ack）接收到的消息
5.  RabbitMQ从队列中删除相应的已经被确认的消息
6.  关闭信道
7.  关闭连接

# AMQP协议 #

AMQP模型就是：生产者将消息发送给交换器，交换器和队列绑定。当生产者发送消息时携带的RoutingKey和绑定的BindingKey匹配的时候，消息被存入相应的队列。

AMQP协议包含三层：

*  Module Layer：位于协议最高层，主要定义一些供客户端调用的命令，客户端可以利用这些命令实现自己的业务逻辑
 *  Session Layer：位于中间层，主要是负责将客户端命令发送给服务端，再将服务端的应答返回给客户端，主要为客户端和服务端之间的通信提供可靠性同步机制和错误处理
 *  Transport Layer：位于最底层，主要传输二进制数据流，提供帧的处理、信道复用、错误检测、数据表示

当客户端与Broker建立连接的时候，会调用factory.newConnection方法，这个方法会进一步封装成Protocol Header的报文头发送给Broker，以此通知Broker本次交互采用的是AMQP协议，紧接着Broker返回connection.start来建立链接，在连接的过程中涉及 connection.start/.Start-OK、Connection.tune/。Tune-OK、Connection.Open/.Open-OK这六个命令的交互。

当客户端调用connection.createChannel方法准备开启信道的时候，其包装Channel.Open命令发送给Broker，等待Channel.Open-OK命令。

当客户端发送消息的时候，需要调用channel.basicPublish方法，对应的AQMP命令为Basic.Publish，注意这个命令和前面涉及的命令有不同，因为这个命令包含了ContentHeader和ContentBody。ContentHeader里面包含的消息体的属性。

当客户端发送消息需要关闭资源时，涉及Channel.Close/.Close-OK与Connection.Close/.Close-OK

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70 1]

消费者客户端同样需要与Broker建立连接，与生产者客户端一样，协议交互同样涉及Connection.Start/ . Start-Ok 、Connection.Tune/.Tune-Ok 和Connection.  
Open/ . Open-Ok

接着在Connection之上建立Channel，和生产者客户端一样，协议涉及Channel.Open/Open-OK

如果在消费之前调用了channel.basicQos(int prefetchCount)的方法来设置消费者客户端最大能够保持的未确认的消息数，那么协议流转会涉及Basic.Qos/.Qos-OK

在真正消费之前，消费者客户端需要向Broker发送Basic.Consume命令将Channel设置为接收模式，之后Broker回执Basic.Consume-OK以告诉消费者客户端准备好消费消息。Broker向消费者客户端推送消息，即Basic.Deliver命令，会携带Content Header 和Content Body。

消费者接收到消息并正确消费之后，向Broker发送确认，即Basic.Ack命令  
消费者停止消费的时候，主动关闭连接，这点和生产者一样，涉及Channel.Close/.Close-OK和Connection.Close/.Close-OK

## AMQP命令 ##

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>是否包含内容体</th> 
   <th>对应客户端中的方法</th> 
   <th>简要描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Connection.Start</td> 
   <td>否</td> 
   <td>factory.newConnection</td> 
   <td>建立连接相关</td> 
  </tr> 
  <tr> 
   <td>Connection.Start-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Connection.Tune</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Connection.Tune-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Connection.Open</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Connection.Open-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Connection.Close</td> 
   <td>否</td> 
   <td>connection.close</td> 
   <td>关闭连接</td> 
  </tr> 
  <tr> 
   <td>Connection.Close-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Channel.Open</td> 
   <td>否</td> 
   <td>connection.openChannel</td> 
   <td>开启信道</td> 
  </tr> 
  <tr> 
   <td>Channel.Open-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Channel.Close</td> 
   <td>否</td> 
   <td>channel.close</td> 
   <td>关闭信道</td> 
  </tr> 
  <tr> 
   <td>Channel.Close-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Exchange.Declare</td> 
   <td>否</td> 
   <td>channel.exchangeDeclare</td> 
   <td>声明交换器</td> 
  </tr> 
  <tr> 
   <td>Exchange.Declare-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Exchange.Delete</td> 
   <td>否</td> 
   <td>channel.exchangeDelete</td> 
   <td>删除交换器</td> 
  </tr> 
  <tr> 
   <td>Exchange.Delete-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Exchange.Bind</td> 
   <td>否</td> 
   <td>channel.exchangeBind</td> 
   <td>交换器和交换器绑定</td> 
  </tr> 
  <tr> 
   <td>Exchange.Bind-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Exchange.Unbind</td> 
   <td>否</td> 
   <td>channel.exchangeUnbind</td> 
   <td>交换器和交换器解绑</td> 
  </tr> 
  <tr> 
   <td>Exchange.Unbind-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Queue.Declare</td> 
   <td>否</td> 
   <td>channel.queueDeclare</td> 
   <td>声明队列</td> 
  </tr> 
  <tr> 
   <td>Queue.Declare-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Queue.Bind</td> 
   <td>否</td> 
   <td>channel.queueBind</td> 
   <td>队列与交换器绑定</td> 
  </tr> 
  <tr> 
   <td>Queue.Bind-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Queue.Purge</td> 
   <td>否</td> 
   <td>channel.queuePurege</td> 
   <td>清除队列中的内容</td> 
  </tr> 
  <tr> 
   <td>Queue.Purge-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Queue.Delete</td> 
   <td>否</td> 
   <td>channel.queueDelete</td> 
   <td>删除队列</td> 
  </tr> 
  <tr> 
   <td>Queue.Delete-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Queue.Unbind</td> 
   <td>否</td> 
   <td>channel.QueueUnbind</td> 
   <td>队列与交换器解绑</td> 
  </tr> 
  <tr> 
   <td>Queue.Unbind-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Basic.Qos</td> 
   <td>否</td> 
   <td>channel.basicQos</td> 
   <td>设置未被确认消费的个数</td> 
  </tr> 
  <tr> 
   <td>Basic.Qos-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Basic-Consume</td> 
   <td>否</td> 
   <td>channel.basicConsume</td> 
   <td>消费消息（推模式）</td> 
  </tr> 
  <tr> 
   <td>Basic-Consume-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Basic-Cancel</td> 
   <td>否</td> 
   <td>channel.basicCancel</td> 
   <td>取消</td> 
  </tr> 
  <tr> 
   <td>Basic.Cancel-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Basic.Publish</td> 
   <td>是</td> 
   <td>channel.basicPublish</td> 
   <td>发送消息</td> 
  </tr> 
  <tr> 
   <td>Basic.Return</td> 
   <td>是</td> 
   <td>无</td> 
   <td>未能成功路由的消息返回</td> 
  </tr> 
  <tr> 
   <td>Basic.Deliver</td> 
   <td>是</td> 
   <td>无</td> 
   <td>Broker推送消息</td> 
  </tr> 
  <tr> 
   <td>Basic.Get</td> 
   <td>否</td> 
   <td>channel.basicGet</td> 
   <td>消费消息（拉模式）</td> 
  </tr> 
  <tr> 
   <td>Basic.Get-OK</td> 
   <td>是</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Basic.Ack</td> 
   <td>否</td> 
   <td>channel.basicAck</td> 
   <td>确认</td> 
  </tr> 
  <tr> 
   <td>Basic.Reject</td> 
   <td>否</td> 
   <td>channel.basicReject</td> 
   <td>拒绝（单条拒绝）</td> 
  </tr> 
  <tr> 
   <td>Basic.Recover</td> 
   <td>否</td> 
   <td>channel.basicRecover</td> 
   <td>请求Broker重新发送未被确认的消息</td> 
  </tr> 
  <tr> 
   <td>Basic.Recover-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Basic.Nack</td> 
   <td>否</td> 
   <td>channel.basicNack</td> 
   <td>拒绝（可批量拒绝）</td> 
  </tr> 
  <tr> 
   <td>Tx.Select</td> 
   <td>否</td> 
   <td>channe.txSelect</td> 
   <td>开启事务</td> 
  </tr> 
  <tr> 
   <td>Tx.Select-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Tx.Commit</td> 
   <td>否</td> 
   <td>channel.txCommit</td> 
   <td>事务提交</td> 
  </tr> 
  <tr> 
   <td>Tx.Commit-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>Tx.Rollback</td> 
   <td>否</td> 
   <td>channel.txRollback</td> 
   <td>事务回滚</td> 
  </tr> 
  <tr> 
   <td>Confirm.Select</td> 
   <td>否</td> 
   <td>channel.confirmSelect</td> 
   <td>开启发送端确认模式</td> 
  </tr> 
  <tr> 
   <td>Confirm.Select-OK</td> 
   <td>否</td> 
   <td>同上</td> 
   <td>同上</td> 
  </tr> 
 </tbody> 
</table>

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70]: /images/20220326/f501d3a64ce84631a2e5e1e71b8ec8cc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70 1]: /images/20220326/209f78c0113c40e4be85360d7dfd2528.png