RabbitMQ练习（AMQP 0-9-1 Overview）

1、What is AMQP 0-9-1

AMQP 0-9-1（高级消息队列协议）是一种网络协议，它允许遵从该协议的客户端（Publisher或者Consumer）应用程序与遵从该协议的消息中间件代理（Broker，如RabbitMQ）进行通信。

AMQP 0-9-1模型的核心概念包括消息发布者（producers/publisher）、消息（messages）、交换机（exchanges）、队列（queues）、绑定（bindings）和消费者（consumers）。

2、Brokers and Their Role

消息代理（Broker）接收来自发布者（发布消息的应用程序，也称为生产者）的消息，并将它们路由到消费者（处理这些消息的应用程序）。

由于它是一种网络协议，发布者、消费者和代理都可以位于不同的机器上。

3、AMQP 0-9-1 Model in Brief

P（Publisher）发布消息必须首先到达X（Exchange），发送的消息带有routing key。
Q（Queue）通过binding key和交换机绑定。
Exchange收到消息后，根据Exchange的类型，处理方式不同。如果是Direct Exchange，消息的routing key和binding key如果匹配，消息则copy到相应的队列。
C（consumer）通过订阅（Subscribe）该队列的消息或者Pull/fetch该队列的消息，该队列的消息最终发往相应的consumer。

队列、交换机和绑定一起组成AMQP实体。（原文：Queues, exchanges and bindings are collectively referred to as AMQP entities.）

4、AMQP 0-9-1 is a Programmable Protocol

AMQP 0-9-1是一个可编程的协议，AMQP 0-9-1实体（实体包括Queues，exchanges and bindings）和消息路由方案主要是由需要使用消息代理的应用程序自己定义的，而不是由消息代理定义。AMQP 0-9-1协议支持声明队列和交换机、定义它们之间的绑定关系、订阅队列等操作，这赋予了应用程序开发者很大的自由度。

5、Exchange类型

Default Exchange：如果发送消息时，不指定Exchange，则该message发送到Default Exchange。Default Exchange类型是Direct Exchange。
Direct Exchange：队列使用binding key和Exchange绑定。Publisher发送携带routing key的消息到Exchange，如果routing key和binding key匹配，消息将copy到该队列。
Fanout Exchange：Fanout Exhcange会把收到的message，发送给所有和它绑定的队列，忽略routing key。
Topic Exchange：和Direct Exchange类似，但binding key是个匹配模板，routing key只要和该模板匹配即可。比如binding key是“#”，表示匹配任何routing key。
Header Exchange：消息Header中的信息，和binding key匹配，忽略rouging key。

Example：“Hello World!”——default exchange:

发布消息未指明exchange，使用default exchange:

Frame 79: 88 bytes on wire (704 bits), 88 bytes captured (704 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:03 (02:42:ac:11:00:03), Dst: 02:42:ac:11:00:02 (02:42:ac:11:00:02)
Internet Protocol Version 4, Src: 172.17.0.3 (172.17.0.3), Dst: 172.17.0.2 (172.17.0.2)
Transmission Control Protocol, Src Port: 37258, Dst Port: 5672, Seq: 382, Ack: 597, Len: 22
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 14Class: Basic (60)Method: Publish (40)ArgumentsTicket: 0Exchange:    <---- no name，使用default exchange，类型为direct exchangeRouting-Key: hello.... ...0 = Mandatory: False.... ..0. = Immediate: False

6、Queues

队列（queues）用于存储由应用程序消费的消息。队列与交换机共享一些属性，但也具有一些额外的属性：

名称（Name）
持久性（Durable），即队列将在代理重启后继续存在。
独占性（Exclusive），即队列只能被一个连接（connection）使用，当该连接关闭时队列将被删除。
自动删除（Auto-delete），即至少有一个消费者订阅过的队列，在最后一个消费者取消订阅时将被删除。
参数（Arguments），这是可选的，被插件和代理特定的特性使用，例如消息存活时间（TTL）、队列长度限制等。

在使用队列之前，必须先声明队列（queue declare）。如果队列尚不存在的话，声明队列将导致创建队列。如果队列已经存在，并且其属性与声明中的属性相同，则声明将不会产生任何效果。

发布者和消费者都可以声明同一个队列，确保消息发布前队列存在。

Example：“Hello World!” —— queue declare

声明创建名称为”hello”的队列。

Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 17Class: Queue (50)Method: Declare (10)ArgumentsTicket: 0Queue: hello.... ...0 = Passive: False.... ..0. = Durable: False.... .0.. = Exclusive: False.... 0... = Auto-Delete: False...0 .... = Nowait: FalseArguments

Example：“Publish/Subscribe” —— 临时队列

Frame 20: 86 bytes on wire (688 bits), 86 bytes captured (688 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:04 (02:42:ac:11:00:04), Dst: 02:42:ac:11:00:02 (02:42:ac:11:00:02)
Internet Protocol Version 4, Src: 172.17.0.4 (172.17.0.4), Dst: 172.17.0.2 (172.17.0.2)
Transmission Control Protocol, Src Port: 40738, Dst Port: 5672, Seq: 388, Ack: 583, Len: 20
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 12Class: Queue (50)Method: Declare (10)ArgumentsTicket: 0Queue:    < -------- no name.... ...0 = Passive: False.... ..0. = Durable: False.... .1.. = Exclusive: True.... 0... = Auto-Delete: False...0 .... = Nowait: FalseArgumentsFrame 21: 117 bytes on wire (936 bits), 117 bytes captured (936 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:02 (02:42:ac:11:00:02), Dst: 02:42:ac:11:00:04 (02:42:ac:11:00:04)
Internet Protocol Version 4, Src: 172.17.0.2 (172.17.0.2), Dst: 172.17.0.4 (172.17.0.4)
Transmission Control Protocol, Src Port: 5672, Dst Port: 40738, Seq: 583, Ack: 408, Len: 51
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 43Class: Queue (50)Method: Declare-Ok (11)ArgumentsQueue: amq.gen-n1_VgX3pWoYEt6CV9STq8A <--- broker给起的临时名字，amp开头Message-Count: 0Consumer-Count: 0

7、Bindings

队列通过绑定键和exchange绑定。exchange收到message以后，根据message携带的routing key和binding key匹配，决定把message从哪个队列发出。

另外，队列声明后创建队列，缺省会以该队列的名字为绑定键和default exchange进行绑定。

Example："Hello World!" ——Binding

1、Consumer声明队列，名字为hello

Frame 16: 91 bytes on wire (728 bits), 91 bytes captured (728 bits) on interface docker0, id 0
Ethernet II, Src: receiver (02:42:ac:11:00:04), Dst: rabbitmq (02:42:ac:11:00:02)
Internet Protocol Version 4, Src: receiver (172.17.0.4), Dst: rabbitmq (172.17.0.2)
Transmission Control Protocol, Src Port: 47890, Dst Port: 5672, Seq: 357, Ack: 571, Len: 25
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 17Class: Queue (50)Method: Declare (10)ArgumentsTicket: 0Queue: hello.... ...0 = Passive: False.... ..0. = Durable: False.... .0.. = Exclusive: False.... 0... = Auto-Delete: False...0 .... = Nowait: FalseArguments

2、rabbitmq创建hello队列，并且通过hello绑定键和default exchage绑定

root@7c53a5705902:/# rabbitmqctl list_queues
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name    messages
hello   0    <-----创建的队列
root@7c53a5705902:/# rabbitmqctl list_exchanges
Listing exchanges for vhost / ...
name    type
amq.topic       topic
amq.fanout      fanout
amq.direct      direct
amq.headers     headers
amq.match       headers
amq.rabbitmq.trace      topicdirect
root@7c53a5705902:/# rabbitmqctl list_bindings
Listing bindings for vhost /...
source_name     source_kind     destination_name        destination_kind        routing_key     argumentsexchange        hello   queue   hello   []  <---绑定键为hello
root@7c53a5705902:/#

3、publisher（sender）发布消息，routing key为hello，未指定exchange（使用default exchange）

Frame 42: 88 bytes on wire (704 bits), 88 bytes captured (704 bits) on interface docker0, id 0
Ethernet II, Src: sender (02:42:ac:11:00:03), Dst: rabbitmq (02:42:ac:11:00:02)
Internet Protocol Version 4, Src: sender (172.17.0.3), Dst: rabbitmq (172.17.0.2)
Transmission Control Protocol, Src Port: 48864, Dst Port: 5672, Seq: 382, Ack: 597, Len: 22
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 14Class: Basic (60)Method: Publish (40)ArgumentsTicket: 0Exchange:  <--- 没有名字，使用default exchangeRouting-Key: hello .... ...0 = Mandatory: False.... ..0. = Immediate: False

8、Consumers

消费者（consumer）可以通过订阅（Subscribe）的方式或者轮询（Polling）的方式，获得某个队列的消息。轮询的方式不推荐，效率比较低。

每个消费者分配有一个consumer tag。

Example：“Hello World” —— Consumer

Packet comments生成Consumer-Tag
Frame 18: 130 bytes on wire (1040 bits), 130 bytes captured (1040 bits) on interface docker0, id 0
Ethernet II, Src: receiver (02:42:ac:11:00:04), Dst: rabbitmq (02:42:ac:11:00:02)
Internet Protocol Version 4, Src: receiver (172.17.0.4), Dst: rabbitmq (172.17.0.2)
Transmission Control Protocol, Src Port: 47890, Dst Port: 5672, Seq: 382, Ack: 597, Len: 64
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 56Class: Basic (60)Method: Consume (20)ArgumentsTicket: 0Queue: helloConsumer-Tag: ctag1.a0f530198eb74e1a85390418796baf1a.... ...0 = No-Local: False.... ..1. = No-Ack: True.... .0.. = Exclusive: False.... 0... = Nowait: FalseFilter

8.1 Message Acknowledgements

AMQP 0-9-1 规范允许消费者控制消息确认机制。有两种确认模式：

在代理（Broker）向应用程序发送消息后（使用 basic.deliver 或 basic.get-ok 方法）。
在应用程序发送回确认后（使用 basic.ack 方法）。

前者被称为自动确认模型（automatic acknowledgement model），而后者被称为显式确认模型（explicit acknowledgement model)。

自动确认模型中，当消息代理将消息发送给应用程序后，如果没有任何错误，消息将自动从队列中删除，无需应用程序显式确认。

显式确认模型中，应用程序选择何时发送确认。它可以在接收到消息后立即发送，或者在将消息持久化到数据库后再进行处理，或者在完全处理完消息后发送（例如，成功获取网页、处理并存储到某个持久化数据库之后）。

Example：“Worker queue” —— 显式确认

1、消费者要求显示确认

Frame 20: 135 bytes on wire (1080 bits), 135 bytes captured (1080 bits) on interface docker0, id 0
Ethernet II, Src: receiver (02:42:ac:11:00:03), Dst: rabbitmq (02:42:ac:11:00:05)
Internet Protocol Version 4, Src: receiver (172.17.0.3), Dst: rabbitmq (172.17.0.5)
Transmission Control Protocol, Src Port: 34932, Dst Port: 5672, Seq: 406, Ack: 614, Len: 69
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 61Class: Basic (60)Method: Consume (20)ArgumentsTicket: 0Queue: task_queueConsumer-Tag: ctag1.0028a0f65f8645efbcf68f81bdbfd278.... ...0 = No-Local: False.... ..0. = No-Ack: False.... .0.. = Exclusive: False.... 0... = Nowait: FalseFilter

2、消费者收到消息后，进行显式确认

Frame 247: 87 bytes on wire (696 bits), 87 bytes captured (696 bits) on interface docker0, id 0
Ethernet II, Src: receiver (02:42:ac:11:00:03), Dst: rabbitmq (02:42:ac:11:00:05)
Internet Protocol Version 4, Src: receiver (172.17.0.3), Dst: rabbitmq (172.17.0.5)
Transmission Control Protocol, Src Port: 34932, Dst Port: 5672, Seq: 475, Ack: 782, Len: 21
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 13Class: Basic (60)Method: Ack (80)ArgumentsDelivery-Tag: 1.... ...0 = Multiple: False

8.2 Prefetching Messages

在多个消费者共享一个队列的情况下，能够指定每个消费者在发送下一个确认之前可以一次性接收多少条消息是非常有用的。这可以作为一种简单的负载均衡技术，或者在消息倾向于批量发布时提高吞吐量。例如，如果一个生产者应用程序由于其工作性质每分钟发送一次消息。

需要注意的是，RabbitMQ仅支持在通道级别设置预取计数（prefetch-count），而不是基于连接或基于大小的预取。

Example：“Worker queue” ——prefetch-count

Frame 18: 85 bytes on wire (680 bits), 85 bytes captured (680 bits) on interface docker0, id 0
Ethernet II, Src: receiver (02:42:ac:11:00:03), Dst: rabbitmq (02:42:ac:11:00:05)
Internet Protocol Version 4, Src: receiver (172.17.0.3), Dst: rabbitmq (172.17.0.5)
Transmission Control Protocol, Src Port: 34932, Dst Port: 5672, Seq: 387, Ack: 602, Len: 19
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 11Class: Basic (60)Method: Qos (10)ArgumentsPrefetch-Size: 0Prefetch-Count: 1.... ...0 = Global: False

9、Message Attributes and Payload

在AMQP 0-9-1模型中，消息具有属性（attributes）。有些属性非常常见，以至于AMQP 0-9-1规范定义了它们，应用程序开发者无需考虑确切的属性名称。一些例子包括：

内容类型（Content type）
内容编码（Content encoding）
路由键（Routing key）
投递模式（Delivery mode），即消息是否持久化
消息优先级（Message priority）
消息发布时间戳（Message publishing timestamp）
消息有效期（Expiration period）
发布应用的ID（Publisher application id）

有些属性由AMQP代理使用，但大多数属性都可以由接收它们的应用程序进行解释。有些属性是可选的，称为头部（headers），它们类似于HTTP中的X-Headers。消息属性在消息发布时设置。

消息还包含一个有效载荷（即它们携带的数据，正文（body）），AMQP代理将其视为不透明的字节数组。代理不会检查或修改有效载荷。消息有可能只包含属性而没有有效载荷。通常使用如JSON、Thrift、Protocol Buffers和MessagePack等序列化格式来序列化结构化数据，以便将其作为消息的有效载荷发布。协议对等方通常使用“content-type”和“content-encoding”字段来传达这些信息，但这仅是基于约定。

Example：“Hello World!”——message组成

Frame 80: 88 bytes on wire (704 bits), 88 bytes captured (704 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:03 (02:42:ac:11:00:03), Dst: 02:42:ac:11:00:02 (02:42:ac:11:00:02)
Internet Protocol Version 4, Src: 172.17.0.3 (172.17.0.3), Dst: 172.17.0.2 (172.17.0.2)
Transmission Control Protocol, Src Port: 37258, Dst Port: 5672, Seq: 404, Ack: 597, Len: 22
Advanced Message Queuing ProtocolType: Content header (2)Channel: 1Length: 14Class ID: Basic (60)Weight: 0Body size: 12Property flags: 0x0000PropertiesFrame 81: 86 bytes on wire (688 bits), 86 bytes captured (688 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:03 (02:42:ac:11:00:03), Dst: 02:42:ac:11:00:02 (02:42:ac:11:00:02)
Internet Protocol Version 4, Src: 172.17.0.3 (172.17.0.3), Dst: 172.17.0.2 (172.17.0.2)
Transmission Control Protocol, Src Port: 37258, Dst Port: 5672, Seq: 426, Ack: 597, Len: 20
Advanced Message Queuing ProtocolType: Content body (3)Channel: 1Length: 12Payload: 48656c6c6f20576f726c6421 <-- "Hello World!"

10、AMQP 0-9-1 Methods

AMQP 0-9-1 被构建为多种方法（Methods）。方法定义了一系列操作（类似于HTTP方法）。方法又被归为不同的类别（Class）。这些类别只是AMQP方法的逻辑分组。

Example：“Hello World!" —— Basic Methods

1、Basic.Consume

Frame 46: 135 bytes on wire (1080 bits), 135 bytes captured (1080 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:04, Dst: 02:42:ac:11:00:05
Internet Protocol Version 4, Src: 172.17.0.4, Dst: 172.17.0.5
Transmission Control Protocol, Src Port: 44490, Dst Port: 5672, Seq: 406, Ack: 614, Len: 69
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 61Class: Basic (60)Method: Consume (20)ArgumentsTicket: 0Queue: task_queueConsumer-Tag: ctag1.9bf1bfaa9e3a499fa3680f0da61e03c6.... ...0 = No-Local: False.... ..0. = No-Ack: False.... .0.. = Exclusive: False.... 0... = Nowait: FalseFilter

2、Basic.Publish

Frame 70: 93 bytes on wire (744 bits), 93 bytes captured (744 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:02, Dst: 02:42:ac:11:00:05
Internet Protocol Version 4, Src: 172.17.0.2, Dst: 172.17.0.5
Transmission Control Protocol, Src Port: 53486, Dst Port: 5672, Seq: 387, Ack: 602, Len: 27
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 19Class: Basic (60)Method: Publish (40)ArgumentsTicket: 0Exchange: Routing-Key: task_queue.... ...0 = Mandatory: False.... ..0. = Immediate: False

3、Basic.Deliver

Frame 75: 183 bytes on wire (1464 bits), 183 bytes captured (1464 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:05, Dst: 02:42:ac:11:00:03
Internet Protocol Version 4, Src: 172.17.0.5, Dst: 172.17.0.3
Transmission Control Protocol, Src Port: 5672, Dst Port: 34932, Seq: 665, Ack: 475, Len: 117
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 64Class: Basic (60)Method: Deliver (60)ArgumentsConsumer-Tag: ctag1.0028a0f65f8645efbcf68f81bdbfd278Delivery-Tag: 1.... ...0 = Redelivered: FalseExchange: Routing-Key: task_queue
Advanced Message Queuing ProtocolType: Content header (2)Channel: 1Length: 15Class ID: Basic (60)Weight: 0Body size: 14Property flags: 0x1000PropertiesDelivery-Mode: 2
Advanced Message Queuing ProtocolType: Content body (3)Channel: 1Length: 14Payload: 4669727374206d6573736167652e

11、Connections

AMQP 0-9-1 连接（connection）通常是长期保持的。AMQP 0-9-1 是一个应用层协议，它使用 TCP 来确保消息的可靠传输。连接时使用认证，并可以通过 TLS（SSL）进行保护。当应用程序不再需要与服务器连接时，应该优雅地关闭其 AMQP 0-9-1 连接，而不是突然关闭底层的 TCP 连接。

Example: "Worker queue” —— connection/channel/vhosts

Frame 6: 587 bytes on wire (4696 bits), 587 bytes captured (4696 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:05, Dst: 02:42:ac:11:00:03
Internet Protocol Version 4, Src: 172.17.0.5, Dst: 172.17.0.3
Transmission Control Protocol, Src Port: 5672, Dst Port: 34932, Seq: 1, Ack: 9, Len: 521
Advanced Message Queuing ProtocolType: Method (1)Channel: 0Length: 513Class: Connection (10)Method: Start (10)ArgumentsVersion-Major: 0Version-Minor: 9Server-Propertiescapabilities (field table)cluster_name (string): rabbit@b1b73bb557e3copyright (string): Copyright (c) 2007-2024 Broadcom Inc and/or its subsidiariesinformation (string): Licensed under the MPL 2.0. Website: https://rabbitmq.complatform (string): Erlang/OTP 26.2.5.2product (string): RabbitMQversion (string): 3.13.6Mechanisms: 414d51504c41494e20504c41494eLocales: 656e5f5553

12、Channels

有些应用程序需要与代理建立多个连接。然而，同时保持许多 TCP 连接是不可取的，因为这样做会消耗系统资源并使配置防火墙变得更加困难。AMQP 0-9-1 通过通道（channel）实现了连接的多路复用，可以将通道视为“共享单个 TCP 连接的轻量级连接”。

客户端执行的每个协议操作都发生在一个通道上。特定通道上的通信与另一个通道上的通信是完全隔离的，因此每个协议方法也携带一个通道 ID（即通道号），代理和客户端都使用这个整数来确定该方法是为哪个通道准备的。

通道只存在于连接的上下文中，而不会单独存在。当连接关闭时，其上的所有通道也会关闭。

Example: "Worker queue” —— connection/channel/vhosts

Frame 14: 79 bytes on wire (632 bits), 79 bytes captured (632 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:03, Dst: 02:42:ac:11:00:05
Internet Protocol Version 4, Src: 172.17.0.3, Dst: 172.17.0.5
Transmission Control Protocol, Src Port: 34932, Dst Port: 5672, Seq: 344, Ack: 555, Len: 13
Advanced Message Queuing ProtocolType: Method (1)Channel: 1Length: 5Class: Channel (20)Method: Open (10)ArgumentsOut-Of-Band:

13、Virtual Hosts

为了让单个代理能够托管多个隔离的“环境”（用户组、交换器、队列等），AMQP 0-9-1 引入了虚拟主机（vhosts）的概念。它们与许多流行的 Web 服务器使用的虚拟主机类似，为 AMQP 实体提供了完全隔离的环境。协议客户端在连接协商期间指定他们想要使用的虚拟主机。

Example: "Worker queue” —— connection/channel/vhosts

Frame 11: 82 bytes on wire (656 bits), 82 bytes captured (656 bits) on interface docker0, id 0
Ethernet II, Src: 02:42:ac:11:00:03, Dst: 02:42:ac:11:00:05
Internet Protocol Version 4, Src: 172.17.0.3, Dst: 172.17.0.5
Transmission Control Protocol, Src Port: 34932, Dst Port: 5672, Seq: 328, Ack: 542, Len: 16
Advanced Message Queuing ProtocolType: Method (1)Channel: 0Length: 8Class: Connection (10)Method: Open (40)ArgumentsVirtual-Host: /   <--表示连接到根路径，通常用于访问AMQP代理的默认虚拟主机。Capabilities: .... ...1 = Insist: True

14、AMQP is Extensible

AMQP 0-9-1是一个高度可扩展的消息队列协议，它提供了多种扩展点来适应不同的应用场景和需求。以下是AMQP 0-9-1的几个关键扩展点：

自定义exchange类型：开发者可以根据特定的路由需求实现自定义的exchange类型，比如基于地理位置数据的路由，这些是AMQP 0-9-1预设exchange类型无法很好覆盖的。
exchange和queue的声明扩展：在声明exchange和队列时，可以包含额外的属性，这些属性可以被消息代理（如RabbitMQ）使用，例如，RabbitMQ中的每个队列消息的TTL（生存时间）就是通过这种方式实现的。
代理特定的协议扩展：AMQP 0-9-1允许代理实现特定的协议扩展，例如RabbitMQ实现的一些扩展。
新方法类的引入：可以引入新的AMQP 0-9-1方法类来支持更多的操作和特性。
代理的插件扩展：代理可以通过插件进行扩展，例如RabbitMQ的管理前端和HTTP API就是作为插件实现的。