TOC

RabbitMQ & AMQP

作者：
日期：
标签： MQ RabbitMQ AMQP

AMQP

Advanced Message Queuing Protocol, 高级消息队列协议

久负盛名的投资公司摩根大通（JPMorgan Chase）在 2005 年前后设计了 AMQP，并和红帽一同采用 Java 实现了这个协议（没过多久就改用 C++ 重构了一遍），这就是后来的 Apache Qpid。后来的一些消息队列也都支持 AMQP 协议，比如 RabbitMQ（采用 Erlang 开发）、Apache ActiveMQ（Java）。

其他常见的 MQ 协议还有：STOMP [^stomp], MQTT [^mqtt]，有时也会和 XMPP [^xmpp] 做对比。

  • 2006/06 版本 0-8
  • 2006/12 版本 0-9
  • 2008/11 版本 0-9-1 RabbitMQ 实现
  • 2011/10 版本 1.0
    • AMQP 移交给 OASIS 组织之后发布的第一个版本
    • 2014/04 该版本成为 ISO 国际标准。

基本概念

AMQP Arch

  • Message
    • DeliveryTag 这个标记十分重要
    • Properties
    • Header
    • Body
    • Content Type
    • Content Encoding
  • Message Queue
  • Message Broker
  • Message-Oriented Middleware 消息中间件，有时简写 MOM
  • Connection TCP 连接, 服务器永远不会主动关闭连接
  • Channel 通道，或者叫信道，逻辑连接，不同通道之间是完全隔离的 (ChannelID)
    • 通道的打开关闭
    • 多线程可以使用各自的通道
  • Server 消息队列服务器，就是指 Broker
  • Virtual Host 虚拟主机，消息队列中的逻辑隔离
  • Publisher 消息生产者
  • Exchange 交换机
  • Queue 队列
  • Binding 绑定，指定了队列和交换机之间的关系
  • RoutingKey 路由键，Binding 的附加参数，对消息进行过滤
    注意：有些地方将绑定时指定的 RoutingKey 叫做 BindingKey
  • Comsumer 消息消费者

基本流程

AMQP 协议中，基本数据单位是帧。有 9 种帧结构用来开启、控制、关闭两点之间的信息传输链路：

  1. 打开（连接）open
  2. 开始（会话）begin
  3. 附加（链路）attach
  4. 传输 transfer
  5. 流量控制 flow
  6. 状态通信 disposition
  7. 分离（链路）detach
  8. 结束（会话）end
  9. 关闭（连接）close

连接 Conection，会话 Session，链路 Link。
链路是单向的数据传输通道，消息（Transfer 帧）就在链路上传输。

+ OpenConnection
|   + StartSession
|   |   + AttachLink
|   |   |     Transfer
|   |   |     Flow
|   |   |     Disposition
|   |   + DetachLink
|   + EndSession
+ CloseConnection
  1. 定义 Exchange、Queue、Binding
  2. 生产者将消息投递给 Exchange
  3. Exchange 根据实现定义的路由策略（Binding）将消息转发到 Queue
  4. 消费者从 Queue 中拿到消息

RabbitMQ

实现了 AMQP 0-9-1。

端口

ps -ef | grep rabbitmq | grep -v grep
rabbitmq  966640       1  5 11:09 ?        00:00:12 /usr/lib/erlang/erts-11.1.8/bin/beam.smp -W w -K true -A 64 -MBas ageffcbf -MHas ageffcbf -MBlmbcs 512 -MHlmbcs 512 -MMmcs 30 -P 1048576 -t 5000000 -stbt db -zdbbl 128000 -- -root /usr/lib/erlang -progname erl -- -home /var/lib/rabbitmq -- -pa  -noshell -noinput -s rabbit boot -boot start_sasl -lager crash_log false -lager handlers []
rabbitmq  966744  966640  0 11:09 ?        00:00:00 erl_child_setup 65536
rabbitmq  966775       1  0 11:09 ?        00:00:00 /usr/lib/erlang/erts-11.1.8/bin/epmd -daemon
rabbitmq  966802  966744  0 11:09 ?        00:00:00 inet_gethost 4
rabbitmq  966803  966802  0 11:09 ?        00:00:00 inet_gethost 4

sudo nmap -p 1-65535 localhost
4369/tcp  open  epmd
5672/tcp  open  amqp
15672/tcp open  unknown
25672/tcp open  unknown
  • 4369 Erlang 端口映射器守护程序 (epmd)
  • 5672 服务端口
  • 15672 Web 接口
  • 25672 不知道干嘛的，来自 inet_dist_listen_min - inet_dist_listen_max, 我本机默认配置

Exchange 与队列

type exchangeDeclare struct {
    reserved1  uint16
    Exchange   string
    Type       string
    Passive    bool
    Durable    bool
    AutoDelete bool
    Internal   bool
    NoWait     bool
    Arguments  Table
}

type queueDeclare struct {
    reserved1  uint16
    Queue      string
    Passive    bool
    Durable    bool
    Exclusive  bool
    AutoDelete bool
    NoWait     bool
    Arguments  Table
}

type queueBind struct {
    reserved1  uint16
    Queue      string
    Exchange   string
    RoutingKey string
    NoWait     bool
    Arguments  Table
}

Exchange 类型

AMQP 中定义的 4 种 Exchange 类型：

  1. direct 需要符合 RoutingKey 的完全匹配
  2. topic 支持模糊匹配：# 代表任意个单词，* 代表一个单词
    • RoutingKey 应该是小数点隔开的单词，另外，不可超过 255 字节。
    • BindingKey 可以包含上面说的模糊匹配字符。
  3. fanout 将消息广播给所有绑定到该 Exchange 的 Queue，忽略 RoutingKey
  4. headers 采用消息的 Header 与 Binding 的属性来进行匹配，忽略 RoutingKey
    • Binding 中 x- 开头的属性不会参与匹配。
    • Binding 可以定义 x-match 属性，any (默认值) 表示匹配中一个字段就行，all 表示所有字段都匹配。
    • 奇怪的是官网教程中没有这种类型的示例。

预置 Exchange:

  • (AMQP default) direct
  • amq.direct direct
  • amq.fanout fanout
  • amq.headers headers
  • amq.match headers
  • amq.rabbitmq.log topic
  • amq.rabbitmq.trace topic
  • amq.topic topic

默认 Exchange (AMQP default) 有点特殊，所有的队列都自动绑定在上面，然后又是 direct 类型，这样一来，生产者 publish 时如果将消息投递到名字为空字符串的这个 Exchange，RoutingKey 填写队列名字，就可以直接将消息投递到队列中。

确认机制

  1. 来自 TCP 的启发。
  2. 分成消费者确认和生产者确认两部分。
  3. DeliveryTag 对于确认机制至关重要。

具体下来就是：

  • Basic.Ack basic_ack(delivery_tag=0, multiple=False) 可以一次确认多个消息
  • Basic.Nack basic_nack(delivery_tag=None, multiple=False, requeue=True)
  • Basic.Reject basic.reject(delivery_tag=None, requeue=True)

Nack 是 RabbitMQ 对 AMQP 的拓展，和 Reject 作用相同，唯一不同的是，Nack 可以一次拒绝该通道之前（小于 delivery_tag）所有没有 ACK 的消息。

关于自动确认模式

命令

Connection

  1. Connection.Start
  2. Connection.StartOk
  3. Connection.Secure
  4. Connection.SecureOk
  5. Connection.Tune
  6. Connection.TuneOk
  7. Connection.Open
  8. Connection.OpenOk
  9. Connection.Close
  10. Connection.CloseOk
  11. Connection.Blocked
  12. Connection.Unblocked

Channel

  1. Channel.Open
  2. Channel.OpenOk
  3. Channel.Flow
  4. Channel.FlowOk
  5. Channel.Close
  6. Channel.CloseOk

Access

  1. Access.Request
  2. Access.RequestOk

Exchange 交换器

  1. Exchange.Declare
  2. Exchange.DeclareOk
  3. Exchange.Delete
  4. Exchange.DeleteOk
  5. Exchange.Bind 交换器绑定 (RabbitMQ 拓展)
  6. Exchange.BindOk
  7. Exchange.Unbind 交换器解绑 (RabbitMQ 拓展)
  8. Exchange.UnbindOk

Queue 队列

  1. Queue.Declare 队列声明
  2. Queue.DeclareOk
  3. Queue.Bind 队列绑定 (到交换器)
  4. Queue.BindOk
  5. Queue.Purge 队列清空 (没分配的，也就是 unack 不会被清空)
  6. Queue.PurgeOk
  7. Queue.Delete 队列删除
  8. Queue.DeleteOk
  9. Queue.Unbind 队列解绑
  10. Queue.UnbindOk

Basic

  1. Basic.Qos Quality of Service 设置：
    1. prefetch_size
    2. prefetch_count
  2. Basic.QosOk
  3. Basic.Consume 消费者开始消费
  4. Basic.ConsumeOk
  5. Basic.Cancel 取消消费者订阅
  6. Basic.CancelOk
  7. Basic.Publish 发布消息
  8. Basic.Return
  9. Basic.Deliver
  10. Basic.Get 直接从指定队列获取消息
  11. Basic.GetOk
  12. Basic.GetEmpty
  13. Basic.Ack 确认
  14. Basic.Reject
  15. Basic.RecoverAsync
  16. Basic.Recover
  17. Basic.RecoverOk
  18. Basic.Nack 负确认 (RabbitMQ 拓展)

Tx 事务

  1. Tx.Select
  2. Tx.SelectOk
  3. Tx.Commit
  4. Tx.CommitOk
  5. Tx.Rollback
  6. Tx.RollbackOk

Confirm

  1. Confirm.Select
  2. Confirm.SelectOk

消息格式

来自 AMQP 协议文件 amqp-0-9-1.xml 中 Basic 的字段定义：

<!--  These are the properties for a Basic content  -->

<!--  MIME typing -->
<field name = "content-type"    domain = "shortstr"   label = "MIME content type" />
<!--  MIME typing -->
<field name = "content-encoding" domain = "shortstr"  label = "MIME content encoding" />
<!--  For applications, and for header exchange routing -->
<field name = "headers"         domain = "table"      label = "message header field table" />
<!--  For queues that implement persistence -->
<field name = "delivery-mode"   domain = "octet"      label = "non-persistent (1) or persistent (2)" />
<!--  For queues that implement priorities -->
<field name = "priority"        domain = "octet"      label = "message priority, 0 to 9" />
<!--  For application use, no formal behaviour -->
<field name = "correlation-id"  domain = "shortstr"   label = "application correlation identifier" />
<!--  For application use, no formal behaviour but may hold the
      name of a private response queue, when used in request messages -->
<field name = "reply-to"        domain = "shortstr"   label = "address to reply to" />
<!--  For implementation use, no formal behaviour -->
<field name = "expiration"      domain = "shortstr"   label = "message expiration specification" />
<!--  For application use, no formal behaviour -->
<field name = "message-id"      domain = "shortstr"   label = "application message identifier" />
<!--  For application use, no formal behaviour -->
<field name = "timestamp"       domain = "timestamp"  label = "message timestamp" />
<!--  For application use, no formal behaviour -->
<field name = "type"            domain = "shortstr"   label = "message type name" />
<!--  For application use, no formal behaviour -->
<field name = "user-id"         domain = "shortstr"   label = "creating user id" />
<!--  For application use, no formal behaviour -->
<field name = "app-id"          domain = "shortstr"   label = "creating application id" />
<!--  Deprecated, was old cluster-id property -->
<field name = "reserved"        domain = "shortstr"   label = "reserved, must be empty" />

来自 rabbitmq/amqp091-go@v1.10.0/types.go 中的消息类型定义：

// Publishing captures the client message sent to the server.  The fields
// outside of the Headers table included in this struct mirror the underlying
// fields in the content frame.  They use native types for convenience and
// efficiency.
type Publishing struct {
    // Application or exchange specific fields,
    // the headers exchange will inspect this field.
    Headers Table

    // Properties
    ContentType     string // MIME content type
    ContentEncoding string // MIME content encoding
    DeliveryMode    uint8  // Transient (0 or 1) or Persistent (2)
    Priority        uint8  // 0 to 9
    CorrelationId   string // correlation identifier
    ReplyTo         string // address to to reply to (ex: RPC)
    // Expiration represents the message TTL in milliseconds. A value of "0"
    // indicates that the message will immediately expire if the message arrives
    // at its destination and the message is not directly handled by a consumer
    // that currently has the capacatity to do so. If you wish the message to
    // not expire on its own, set this value to any ttl value, empty string or
    // use the corresponding constants NeverExpire and ImmediatelyExpire. This
    // does not influence queue configured TTL values.
    Expiration string
    MessageId  string    // message identifier
    Timestamp  time.Time // message timestamp
    Type       string    // message type name
    UserId     string    // creating user id - ex: "guest"
    AppId      string    // creating application id

    // The application specific payload of the message
    Body []byte
}
发布于码厩技术博客的所有文章，除注明转载外，均为作者原创，欢迎转载，但必须注明出处。
尊重他人劳动，共创开源社区！转载请注明以下信息：
转载来源： 码厩技术博客 []
原文标题：RabbitMQ & AMQP
原文地址：
知识共享许可协议
如果你有魔法，你可以看到一个评论框～