浅谈RabbitMQ的基石—高级消息队列协议（AMQP）

点击上方蓝色字体&＃xff0c;选择“设为星标”

回复”资源“获取更多资源

大数据技术与架构

点击右侧关注&＃xff0c;大数据开发领域最强公众号&＃xff01;

大数据真好玩

点击右侧关注&＃xff0c;大数据真好玩&＃xff01;

  前言

自从去年做了不少流式系统&＃xff08;Flink也好&＃xff0c;Spark Streaming也好&＃xff09;对接RabbitMQ的实时作业。之前一直都在Kafka的领域里摸爬滚打&＃xff0c;对RabbitMQ只是有浅薄的了解而已。随着自己逐渐把RabbitMQ的官方文档大致翻完&＃xff0c;了解到它是高级消息队列协议&＃xff08;Advanced Message Queuing Protocol, AMQP&＃xff09;的一种标准实现。也就是说&＃xff0c;搞清楚AMQP是掌握好RabbitMQ哲学的基础。

当前AMQP的最新版本为1.0&＃xff0c;而主要使用的&＃xff08;也是RabbitMQ实现的&＃xff09;版本为0-9-1。这两个版本之间的差别非常大&＃xff0c;本文抄录的是AMQP 0-9-1的部分细节。

AMQP及其模型

通俗地讲&＃xff0c;AMQP是一个专门为消息中间件设计的、开放标准的应用层协议&＃xff0c;它规定了消息系统中三大组件——消息服务器/代理节点&＃xff08;server/broker&＃xff09;、生产者/发布者&＃xff08;producer/publisher&＃xff09;、消费者/订阅者&＃xff08;consumer/subscriber&＃xff09;之间的通信规范&＃xff0c;以及代理节点的设计规范等。

AMQP采用的模型就叫做高级消息队列模型&＃xff0c;即AMQ模型&＃xff0c;它的组成可以用下面的简图来表示。

下面就图中出现的一些名词进行解释。

• 交换器&＃xff08;exchange&＃xff09;&＃xff1a;负责将生产者发来的消息按照特定的路由关键字&＃xff08;routing key&＃xff09;投递到相应的队列。

• 队列&＃xff08;queue&＃xff09;&＃xff1a;代理节点中存储将要被消费的消息的载体。

• 绑定&＃xff08;binding&＃xff09;&＃xff1a;交换器与队列之间的映射关系&＃xff0c;可以理解为消息的路由规则。

• AMQP实体&＃xff08;AMQP entity&＃xff09;&＃xff1a;交换器、队列和绑定三者合起来就称为一个AMQP实体&＃xff0c;图中未示出。交换器、队列和绑定都可以有一个或多个。

• 虚拟主机&＃xff08;virtual host&＃xff09;&＃xff1a;在代理节点上逻辑划分的隔离的环境&＃xff0c;其内部包含一个或多个AMQP实体&＃xff0c;且虚拟主机之间互不影响。虚拟主机可以复用节点&＃xff0c;并实现权限管理和多租户。

• 连接&＃xff08;connection&＃xff09;&＃xff1a;发布者、消费者与代理节点之间建立的连接&＃xff0c;为了保证可靠性&＃xff0c;一般都是TCP长连接。

• 通道&＃xff08;channel&＃xff09;&＃xff1a;对连接的轻量级复用&＃xff0c;主要针对多线程的发布者、消费者&＃xff0c;因为建立多个TCP连接是很贵的操作&＃xff0c;频繁建立和销毁连接也是不科学的。

接下来对交换器和队列这两个比较重要的组件进行介绍&＃xff0c;顺便牵出一些其他的东西。

交换器

交换器在AMQP实体中负责消息路由。它的路由目的地除了由用户设置的绑定规则来决定之外&＃xff0c;还与交换器的类型有关。AMQP定义了几种默认的交换器。

• 直连交换器&＃xff08;direct exchange&＃xff09;

直连交换器非常简单&＃xff0c;它检查绑定关键字&＃xff08;binding key&＃xff09;与路由关键字&＃xff08;routing key&＃xff09;&＃xff0c;只要两者相同&＃xff0c;即进行投递。

• 扇出交换器&＃xff08;fanout exchange&＃xff09;

扇出交换器比直连交换器更简单&＃xff0c;它会直接将消息路由到所有与它绑定的队列中。

• 主题交换器&＃xff08;topic exchange&＃xff09; 此主题非彼&＃xff08;对就是Kafka里的&＃xff09;主题&＃xff0c;而更类似wildcard matching。具体来讲&＃xff0c;绑定关键字是由多个域组成的点号分隔的字符串&＃xff0c;每个域可以是实际的单词&＃xff0c;也可以是通配符&＃xff0c;如星号 " * " 表示一个词&＃xff0c;"#" 表示0个或多个词。在实际路由时&＃xff0c;根据路由关键字与绑定关键字的匹配结果来投递。比如在下图中&＃xff0c;带有"little.C.magic"关键字的消息会投递到队列1&＃xff0c;而带有"bla.bla.B"关键字的消息会投递到队列2。

• 头部交换器&＃xff08;header exchange&＃xff09; AMQP消息与HTTP报文的格式类似&＃xff0c;都有头部&＃xff08;header&＃xff09;和消息体&＃xff08;body&＃xff09;&＃xff0c;其中头部会保存与消息相关的许多元数据&＃xff0c;消息体才是有效的载荷&＃xff08;payload&＃xff09;。头部交换器就不依赖绑定关键字和路由关键字的匹配&＃xff0c;而是检查消息头部中的元数据是否匹配&＃xff0c;相对而言更加灵活。

根据AMQP的规定&＃xff0c;交换器的几个重要属性有&＃xff1a;

• 名称&＃xff08;name&＃xff09;&＃xff1b;

• 持久性&＃xff08;durable&＃xff09;&＃xff1a;当代理节点或虚拟主机重置后&＃xff0c;交换器是被保留还是被删除&＃xff1b;

• 自动删除&＃xff08;auto-delete&＃xff09;&＃xff1a;是否在所有队列的绑定解除之后被删除&＃xff1b;

• 扩展参数&＃xff08;arguments&＃xff09;。

如果交换器无法将消息路由到队列该怎么办呢&＃xff1f;AMQP给出了几种解决方法&＃xff0c;一是直接丢弃&＃xff0c;二是返还给生产者&＃xff0c;三是放入死信队列中等待进一步处理。这由消息头部中的属性来决定。

队列和消息

队列相对而言比较简单&＃xff0c;它的主要功能就是存储要被消费的消息。队列也有一些重要的属性&＃xff0c;如下&＃xff1a;

• 名称&＃xff08;name&＃xff09;&＃xff1b;

• 持久性&＃xff08;durable&＃xff09;&＃xff1a;当代理节点或虚拟主机重置后&＃xff0c;队列是被保留还是被删除&＃xff1b;

• 独占性&＃xff08;exclusive&＃xff09;&＃xff1a;是否只允许被一个连接使用&＃xff1b;

• 自动删除&＃xff08;auto-delete&＃xff09;&＃xff1a;是否在所有消费者取消订阅之后被删除&＃xff1b;

• 扩展参数&＃xff08;arguments&＃xff09;&＃xff1a;如队列缓存长度、消息TTL等。

需要注意&＃xff0c;如果一个队列是持久的&＃xff0c;那么只是代表重启之后这个队列不用重新创建而已&＃xff0c;但其中的消息还是有可能被删除。只有那些被标记为persistent的消息才不会被删除。

AMQP规范下的队列和消费者都同时支持推模式和拉模式消费。前者即AMQP实体将消息投递到消费者&＃xff0c;后者即消费者主动地从队列中获取消息。无论推模式还是拉模式&＃xff0c;每个消费者也有一个标识&＃xff0c;称为tag。

在队列中的消息投递出去之后&＃xff0c;消费者需要告诉代理节点自己是否收到了它&＃xff0c;因此会涉及消息确认&＃xff08;ack&＃xff09;的问题。AMQP默认定义了两种ack机制&＃xff1a;

• 自动ack&＃xff1a;当消息从队列中出去后就删除它&＃xff08;即at most once&＃xff09;&＃xff1b;

• 显式ack&＃xff1a;当消费者发送的确认回执到达代理节点后&＃xff0c;再从队列中删除它。如果ack超时&＃xff0c;则会再次尝试投递&＃xff08;即at least once&＃xff09;。

除了ack之外&＃xff0c;消费者在处理时有可能会出现问题&＃xff0c;或认为此消息非法&＃xff0c;因此也会出现拒绝消息&＃xff08;reject&＃xff09;的情况。此时代理节点可以销毁这条消息&＃xff0c;也可以重新将它放入队列并投递给另一个消费者。

vs Kafka&＃xff1f;

说了这么多&＃xff0c;那么Kafka和AMQP有什么关系呢&＃xff1f;答案是没关系。

也就是说&＃xff0c;Kafka不是消息队列。按官方说法&＃xff0c;Kafka是一个流式处理平台&＃xff08;stream processing platform&＃xff09;。Kafka在设计之初是为了支持高吞吐量的日志处理的&＃xff0c;只不过它恰好也可以实现消息队列的大部分功能而已。Kafka所用的“黑科技”&＃xff08;如零拷贝/内存映射&＃xff0c;以及对page cache的利用&＃xff09;都是脱离标准消息队列的设计范畴的&＃xff0c;所以不能简单地认为Kafka比RabbitMQ等符合AMQP的消息队列更优。例如&＃xff0c;RabbitMQ支持死信队列、延迟队列、优先队列、多租户、推模式消费等&＃xff0c;Kafka统统不支持。

版权声明&＃xff1a;

本文为大数据技术与架构整理&＃xff0c;原作者独家授权。未经原作者允许转载追究侵权责任。

编辑&＃xff5c;冷眼丶

微信公众号&＃xff5c;import_bigdata

欢迎点赞&＃43;收藏&＃43;转发朋友圈素质三连

文章不错&＃xff1f;点个【在看】吧&＃xff01; ????