文章目录

• AMQP协议介绍
• AMQP与JMS
• AMQP核心组成
• • Producer(生产者)
  • ConnectionFactory(连接工厂)
  • Connection(连接)
  • 通道
  • 经纪人(中级)
  • VirtualHost(虚拟主机)
  • Exchange(交换机)
  • • 默认交换机
    • 直连交换机
    • 扇型交换机
    • 主题交换机
    • 头交换机
    • 交换机小结
  • 绑定
  • Routing Key(路由键)
  • Queue(队列)
  • Consumer(消费者)
  • 消息
• AMQP工作过程
• AMQP消息机制
• • 消息确认
  • 拒绝消息

AMQP协议介绍

AMQP（Advanced Message Queuing Protocol）高级消息队列协议，提供统一消息服务的标准层协议,是一个开放的应用程序层协议标准,为面向消息的中间件设计。AMQP是一个网络协议,它在进程之间传递异步消息。

基于此协议的客户端和消息中介人可以发送消息,不受客户端/中介人不同的产品、不同的开发语言等限制。

AMQP的主要功能是面向消息、面向队列、路由(包括点到点和发布/订阅)、可靠性和安全性。AMQP强制了消息提供者和客户端的行为,允许不同供应商之间的实际互操作。

AMQP与JMS

JMS是早期试图标准化消息中间体的尝试,它只在API一级标准化,远远没有创造互操作性。

与JMS不同,AMQP是一个Wire-level协议,它描述网络上以字节形式传输的数据的格式为流,因此任何符合这个数据格式的工具,创建和解释消息,可以与其他兼容的工具交互。

AMQP核心组成

Producer(生产者)

生产消息。

ConnectionFactory(连接工厂)

制造连接的工厂。

Connection(连接)

连接,应用程序通过TCP/IP/三个握手和四个波向 Broker 网络连接。

AMQP连接通常是长连接。AMQP是一个基于TCP的应用程序层协议,提供可靠的传输。AMQP使用认证机制,并提供TLS(SSL)保护。当应用程序不再需要连接AMQP代理人时,需要雅致的释放AMQP连接,而不是直接关闭TCP连接.

通道

网络信道，是建立在Connection连接之上的一种轻量级的连接。几乎所有的操作都是在海峡进行的,信道是读写信息的信道,客户可以设置渠道,每个频道代表一个会议任务。

如果把Connection比作一条光纤电缆的话，那么Channel信道就比作成光纤电缆中的其中一束光纤。一个Connection上可以创建任意数量的Channel。

我们的大部分业务都是在通道接口中进行的,包括:

• 队列的声明queueDeclare
• 交换机的声明exchangeDeclare
• 队列绑定队列
• 发布消息basicPublish
• 消费者新闻 BasicConsume等。

经纪人(中级)

接受客户端连接实现AQMP实体服务, 例如rabbitmq.

VirtualHost(虚拟主机)

一个虚拟主机,用于逻辑分离,可以有多个交换机和队列,因为一个虚拟主机的原因,以及一个不能在同一虚拟主机内部拥有相同的名称的交换机。

在单个代理人上实现多个隔离环境(用户、用户组、开关、队列等),AMQP 提供了一个虚拟主机（virtual hosts - vhosts）的概念。这与Web服务器虚拟主机的概念非常相似,这为AMQP实体提供了完全孤立的环境。当连接建立时,AMQP客户端指定要使用哪个虚拟主机。

Exchange(交换机)

通过路由键接收消息并发送消息到绑定队列(没有存储消息的能力)的开关。

交换机是用来发送消息的 AMQP 实体。交换机拿到一个消息之后将它路由给一个或零个队列。它使用哪种路由算法是由交换机类型和绑定（Bindings）规则所决定的。

交换机类型：

• Direct exchange（直连交换机）
• Fanout exchange（扇形交换机）
• Topic exchange（主题交换机）
• Headers exchange（头交换机）

交换机属性：

• 名称:交换名称
• Durability：持久化标志，表明此交换机是否是持久化的
• Auto-delete：删除标志，表明当所有队列在完成使用此exchange时，是否删除
• 论点:依赖代理人本身

交换机状态：

• 持久性
• 暂存（transient）

在消息代理(经纪人)重新启动后仍然存在持续的交换机会,而临时交换则没有(在代理重新启动后需要重新宣布)。

默认交换机

默认交换机（default exchange）实际上是一个由消息代理预先声明好的，没有名字（名字为空字符串）的直连交换机（direct exchange）。

可以认为默认开关是一个特殊的直接连接开关.
默认开关名称:空字符串(AMQP默认)
默认开关类型:直接开关

在创建一个Queue时，只要没指定需要绑定的交换机，都会自动绑定到默认交换机上，绑定的路由键（routing key）名称与队列名称相同。

直连交换机

直连型交换机是根据消息携带的路由键（routing key）将消息投递给对应绑定键的队列。直连交换机用来处理消息的单播路由。

在创建一个Queue时，如果绑定的是直连交换机，可以不必指定routing key的名字，因为他会有一个默认的routing key名称，名称同Queue名称。

直接到网络交换机的队列通常是一个循环分配给多个用户(我们称之为查询)。

工作流程：

1. 将一个队列绑定到某个交换机上时，赋予该绑定一个绑定键（Binding Key），假设为R；
2. 当一个携带着路由键（Routing Key）为R的消息被发送给直连交换机时，交换机会把它路由给绑定键为R的队列。

扇型交换机

扇型交换器将消息路径绑定到所有队列上,而不考虑绑定路径键。

如果N个队列绑定到某个扇型交换机上，当有消息发送给此扇型交换机时，交换机会将消息的拷贝分别发送给这所有的N个队列。扇型交换机一般用来处理消息的广播路由（broadcast routing）。

应用场景：

• 广播消息；
• 群聊功能。

主题交换机

主题交换机是根据routing key和Exchange的类型将message发送到一个或者多个Queue中，我们经常拿他来实现各种publish/subscribe，即发布订阅。

直连交换机的路由规则是严格意义上的匹配，换言之Routing Key必须与Binding Key相匹配的时候才将消息传送给Queue。
主题交换器的路由规则是模糊匹配,可以通过满足规则的一部分的路由器传输。

它约定：

• binding key中可以存在两种特殊字符 * 与#，用于做模糊匹配。其中*用于匹配一个单词，#用于匹配多个单词（可以是零个）
• routing key为一个点号分隔的字符串（我们将被点号分隔开的每一段独立的字符串称为一个单词）

• 当制造商发送消息时路由键=A.A.当A满足时,它满足A.*.*只将路由到队列1;
• 当制造商发送消息 路由键=A.B.当A满足时当A满足时A.*.*和*.B.*将被调到队列1、队列2;
• 当生产者发送消息Routing Key=A.B.C的时候，这时候满足A.*.*和*.B.*和*.*.C它将被引导到队列1、队列2和队列3。

应用场景：

• 有关分类或标签的新闻更新;
• 一个由多个工人完成的备份任务,每个工人负责处理某些特定的任务。

头交换机

头交换机不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的 headers 属性进行匹配。

头交换机可以视为直连交换机的另一种表现形式。但直连交换机的路由键必须是一个字符串，而头属性值则没有这个约束，它们甚至可以是整数或者哈希值（字典）等。灵活性更强（但实际上我们很少用到头交换机）。

工作流程：

1. 绑定一个队列到头交换机上时，会同时绑定多个用于匹配的头（header）；
2. 消息附有头条,并且将有一个"x-match"参数。当"x-match"设置为"any",匹配的消息头条的任何值都可以满足条件,当"x-match"设置为"all",所有为消息标题所需的值都成功匹配。

交换机小结

交换机类型

交换机名称

Binding Key

Routing Key

路由规则

默认交换机

空串，不可修改

不能更改队列的名称

绑定键名称

Routing Key==Binding Key，严格匹配

直连交换机

自定义

默认为队列名可以修改

绑定键名称

Routing Key==Binding Key，严格匹配

扇形交换机

自定义

没有绑定键

没有路由密钥

没有绑定键，自动路由到交换机绑定的所有Queue中

主题交换机

自定义

自定义

自定义

Routing Key==Binding Key，模糊匹配

头交换机

自定义

自定义

自定义

根据发送的消息内容匹配标题属性

绑定

交换与队列之间的虚拟连接。

BindingKey是Exchange和Queue绑定的规则描述。Binding Key指定当前Exchange下，什么样的Routing Key会被下派到当前绑定的Queue中。

Routing Key(路由键)

路由规则,虚拟机可以使用它们来确定如何路由特定消息。

Binding Key与Routing Key的关系

• Binding Key是队列和交换机之间的绑定key；
• 路由密钥是制造商向交换机发送的信息;
• 当Binding Key和Routing Key能对应上时，将该消息放到相应的队列中。

Binding Key是Exchange和Queue绑定的规则描述，这个描述被用来分析Xchange接收消息时,Exchange接收到的消息会带有Routing Key这个字段，Exchange就是根据这个Routing Key和当前Exchange所有绑定的Binding Key做匹配，如果满足要求，就往Binding Key所绑定的Queue发送消息。

各种交换机中的Binding Key与Routing Key

• 默认交换机：Binding Key为Queue名称，不可自定义；Routing Key也为Queue名称时才能成功路由到队列
• 直连交换机：Binding Key为Queue名称，可以自定义；Routing Key于Binding Key相同时才能成功路由到队列
• 扇形交换机：没有绑定键；当然也没有路由密钥。自动路由到交换机绑定的所有Queue中
• 主题交换机：自定义Binding Key；自定义Routing Key。Routing Key==Binding Key，模糊匹配成功才能成功路由到队列
• 头交换机：没有绑定键；当然也没有路由密钥。根据发送的消息内容匹配标题属性

Queue(队列)

存储即将应用的消费信息。

队列属性：

• Name：队列名称
• 持久:消息代理重新启动后仍存在队列
• Exclusive：只被一个连接（connection）使用，而且当连接关闭后队列即被删除
• Auto-delete：当最后一个消费者退订后即被删除
• 参数:一些消息代理人使用他来完成与TTL相似的某些额外的功能

队列创建：
队列不能在声明之后使用.如果队列不存在,声明队列将创建它。如果声明队列已经存在,并且属性完全相同，因此,该声明不会对最初的队列产生任何影响。如果声明中的属性与现有队列中的属性不同,那么一个错误代码为 406 的通道级异常就会被抛出。

队列持久化：
持久化队列会被存储在磁盘上，当消息代理（broker）重启的时候，它依旧存在。没有被持久化的队列称作暂存队列（Transient queues）。并不是所有的场景和案例都需要将队列持久化。

持久化的队列并不会使得路由到它的消息也具有持久性。倘若消息代理挂掉了，重新启动，那么在重启的过程中持久化队列会被重新声明，无论怎样，只有经过持久化的消息才能被重新恢复。

Consumer(消费者)

消费者消费消息。在AMQP中，消费者获取待消费消息的途径有两种：

• 向消费者发送消息的消息中间体(push API)
• 消费者主动获取消息 (pull API)

注:当多个消费者听同一队列时,队列中的消息只由一个消费者使用(并非每个消费者都会一次消费）

消息

消息，服务与应用程序之间传送的数据，由Properties（属性）和body（主体）组成。

属性是对消息进行修饰，比如消息的优先级，延迟等高级特性，主体则就是消息体的内容。

消息属性：

• Content type（内容类型）
• Content encoding（内容编码）
• Routing key（路由键）
• Delivery mode (persistent or not)
• 提交模式(永久或非永久)
• Message priority（消息优先权）
• Message publishing timestamp（消息发布的时间戳）
• Expiration period（消息有效期）
• Publisher application id（发布应用的 ID）

消息主体：
除了这些属性外,AMQP消息还包含一个有效的负载负载(信息实际携带的数据),该数据被AMQP代理处理为一个不透明的字节数。

消息代理不会检查或修改有效负载.消息只能包含属性而不带有有效的负载.它通常使用与JSON相似的格式化数据序列,为了节省，协议缓冲器和MessagePack将结构化数据序列化，将以有效消息载荷的形式发布。AMQP及其同行者们通常使用 “content-type” 和 “content-encoding” 这两个字段来与消息沟通进行有效载荷的辨识工作，但这只是基于协议。

消息持久化：
消息可以以持久的方式发布,AQMP代理将消息存储在磁盘上。 如果服务器重新启动,系统将确认收到的持久消息不会丢失。

简单地将消息发送给一个持久化的交换机或者路由给一个持久化的队列，并不会使得此消息具有持久化性质：消息的持久性完全取决与消息本身的持久模式（persistence mode）。

当消息以持久的方式发布时, 它对性能有一定的影响.

AMQP工作过程

• 发布者（Publisher）发布消息（Message），经由交换机（Exchange）。

• 交换机根据路由规则将接收到的消息分发到与交换机绑定的队列中。

• 最后,AQMP代理人将消息发送给已订阅这个队列的客户或自己获取的客户。

AMQP消息机制

消息确认

消费者在处理消息的时候偶尔会失败或者有时会直接崩溃掉。而且网络原因也有可能引起各种问题。
这给了我们一个问题: AMQP代理在删除消息时是正确的。

AMQP有两个消息确认模式:

• 自动确认模式:消息由消息中介人发送给消费者后立即删除(使用AQMP方法: basic.deliver或 basic。
• 显式确认模式：待消费者发送一个确认回执（acknowledgement）后再删除消息。（使用AMQP方法：basic.ack）

如果一个消费者没有发送确认撤销,则AMQP代理人将重新转发该消息给另一个消费者。 如果当时没有可用的消费者,则该消息代理人将等待下一个消费者登记在这一队列,然后再尝试再次交付。

拒绝消息

当一个消费者接收到某条消息后，处理过程有可能成功，有可能失败。消费者可以向消息代理（消息中间件）表明，本条消息由于 “拒绝消息” 的原因处理失败了（或者未能在此时完成）。
当消息被拒绝时,消费者可以告诉消息代理如何处理消息--销毁它或重新进入队列。

当此队列只有一个消费者时，请确认不要由于拒绝消息并且选择了重新放入队列的行为而引起消息在同一个消费者身上无限循环的情况发生。

在AMQP 中，basic.reject 方法用来执行拒绝消息的操作。但 basic.reject 有个限制：你不能使用它决绝多个带有确认回执（acknowledgements）的消息。但是如果你使用RabbitMQ,那么你可以使用被称作 negative acknowledgements（也叫 nacks）的 AMQP 0-9-1 扩展来解决这个问题。

先自我介绍一下，他高中毕业了13年,曾经在小公司待过，去过华为OPPO等大厂，18年进入阿里，直到现在。了解大多数年轻的Java工程师,想要升技能，经常需要找到自己的成长或向班上汇报。但对于培训机构来说,学费大约是人民币,着实压力不小。当你不在系统时,自我学习的效率很低,而且很持久。也很容易停止天花板技术。所以我为你收集了一个"java开发工具"初衷也很简单，这是一个想帮助自己学习的朋友,却不知道该从哪里学习。同时减少每个人的负担.添加下方名片，你可以得到完整的学习信息

本文由 在线网速测试 整理编辑，转载请注明出处，原文链接：https://www.wangsu123.cn/news/31148.html。