文章

发布者

发表于
作者 jamespud
41 分钟阅读
发布者

发布者

原文

版本：4.0

发布者

概述

本指南涵盖了与发布者相关的各种主题：

  1. 基础知识
  2. 发布者生命周期
  3. 协议差异
  4. 消息属性和投递元数据
  5. 发布者端的数据安全主题（连接恢复、发布者确认）
  6. 异常处理
  7. 资源警报的影响
  8. 不可路由消息的处理
  9. 与发布者相关的指标
  10. 并发考虑因素
  11. 如何临时阻塞所有发布者
  12. 如何排查发布者的常见问题等。

本指南主要聚焦于AMQP 0-9-1协议，并提及了与RabbitMQ支持的其他协议（AMQP 1.0、MQTT和STOMP）在关键协议方面的差异。

术语

“发布者”一词在不同的上下文中含义有所不同。一般来说，在消息传递领域，发布者（也称为“生产者”）是一个发布（产生）消息的应用程序（或应用程序实例）。同一个应用程序也可以消费消息，因此同时也可以是一个消费者。

消息传递协议中也有用于消息投递的持久订阅概念。“订阅”是描述此类实体的常用术语之一，“消费者”是另一个常用术语。RabbitMQ支持的消息传递协议同时使用这两个术语，但RabbitMQ文档更倾向于使用后者。

基础知识

RabbitMQ是一个消息代理。它接受来自发布者的消息，进行路由，如果有可路由到的队列，则存储这些消息以供消费；如果有在线消费者，则立即将消息投递到消费者。

发布者发布消息的目的地因协议而异。在AMQP 0-9-1中，发布者向交换器发布消息。在AMQP 1.0中，消息发布发生在一个链接上。在MQTT中，发布者向主题发布消息。最后，STOMP支持多种目的地类型：主题、队列、AMQP 0-9-1交换器。在协议差异部分将对此进行更详细的介绍。

一条发布的消息必须被路由到一个队列（主题等）。该队列（主题）可能有在线消费者。当消息成功路由到队列，并且有在线消费者可以接受更多投递时，消息将被发送给消费者。

尝试向不存在的队列（主题）发布消息将导致通道级异常，错误代码为404 Not Found，并会关闭尝试发布消息的通道。

发布者生命周期

发布者通常长时间运行：也就是说，在发布者的整个生命周期中，它会发布多条消息。为了发布一条消息而打开一个连接或通道（会话）并不是最优选择。

发布者通常在应用程序启动时打开它们的连接。它们的存活时间通常与连接甚至应用程序的运行时间一样长。

发布者也可以更加动态，在系统事件发生时开始发布消息，当不再需要时停止发布。这在通过Web STOMP和Web MQTT插件使用的WebSocket客户端、移动客户端等场景中很常见。

协议差异

RabbitMQ支持的每种协议在消息发布过程上都非常相似。所有四种协议都允许用户发布带有有效负载（消息体）和一个或多个消息属性（消息头）的消息。

所有四种协议也都支持发布者的确认机制，该机制允许发布应用程序跟踪哪些消息已被代理成功接受，哪些没有，从而继续发布下一批消息或重试发布当前消息。

这些协议的差异通常更多地体现在使用的术语上，而不是语义上。消息属性也因协议而异。

AMQP 0-9-1

在AMQP 0-9-1中，消息在通道上发布到交换器。交换器使用通过在一个或多个队列与交换器之间，或者源交换器与目标交换器之间定义绑定来设置的路由拓扑。成功路由的消息将存储在队列中。

每个实体的作用在AMQP 0-9-1概念指南中有介绍。

发布者确认是发布者的确认机制。

有几种常见的发布者错误类型，通过不同的协议特性来处理：

  1. 发布到不存在的交换器会导致通道错误，该错误会关闭通道，使得在该通道上不能再进行进一步的发布（或任何其他操作）。
  2. 当发布的消息无法路由到任何队列时（例如，因为目标交换器没有定义绑定），如果发布者将mandatory消息属性设置为false（这是默认值），消息将被丢弃或重新发布到备用交换器（如果有的话）。
  3. 当发布的消息无法路由到任何队列，并且发布者将mandatory消息属性设置为true时，消息将被返回给发布者。发布者必须设置一个返回消息处理程序来处理返回的消息（例如，通过记录错误或尝试使用不同的交换器重新发布）。

    AMQP 1.0

    在AMQP 1.0中，消息发布发生在链接的上下文中。

    MQTT

    在MQTT中，消息在连接上发布到一个主题。服务器端的MQTT连接进程通过主题交换器将消息路由到队列。

当发布者选择使用QoS 1时，RabbitMQ会使用PUBACK数据包确认已发布的消息。

发布者可以向服务器提供一个提示，表明发布到该主题的消息必须被保留（存储以便将来投递到新的订阅者）。每个主题只能保留最新发布的消息。

MQTT 5.0的PUBACK数据包包含一个原因代码，用于告知发布者发布是否成功。RabbitMQ返回的原因代码包括：

  1. 0 - Success：消息被成功路由到的所有队列都接受了该消息。
  2. 16 - No matching subscribers：RabbitMQ无法将消息路由到任何队列（因为没有为主题交换器定义绑定）。
  3. 131 - Implementation specific error：RabbitMQ拒绝了该消息（例如，当目标经典队列不可用时）。

在MQTT 3.1和3.1.1中，除了关闭连接外，服务器没有其他机制可以将发布错误传达给客户端。

有关更多信息，请参阅MQTT和MQTT-over-WebSockets指南。

STOMP

STOMP客户端在连接上向一个或多个目的地发布消息，在RabbitMQ的情况下，这些目的地可能具有不同的语义。

STOMP提供了一种服务器将消息处理错误传达回发布者的方式。它的发布者确认变体称为收据，这是客户端在发布时启用的一项功能。

有关更多信息，请参阅STOMP指南、STOMP-over-WebSockets和STOMP 1.2规范。

路由

AMQP 0-9-1

AMQP 0-9-1中的路由由交换器执行。交换器是命名的路由表。表项称为绑定。在AMQP 0-9-1概念指南中有更详细的介绍。

有几种内置的交换器类型：

  1. 主题（Topic）
  2. 扇出（Fanout）
  3. 直连（包括默认交换器）
  4. 头（Headers）

前三种类型在教程中有示例介绍。

插件可以提供更多的交换器类型。一致性哈希交换器、随机路由交换器、内部事件交换器和延迟消息交换器是随RabbitMQ一起发布的交换器插件。与所有插件一样，必须先启用它们才能使用。

不可路由消息的处理

客户端可能会尝试向不存在的目的地（交换器、主题、队列）发布消息。本节介绍不同协议在处理此类情况时的差异。

RabbitMQ收集并公开了一些指标，可用于检测发布不可路由消息的发布者。

AMQP 0-9-1

当发布的消息无法路由到任何队列时（例如，因为目标交换器没有定义绑定），如果发布者将mandatory消息属性设置为false（这是默认值），消息将被丢弃或重新发布到备用交换器（如果有的话）。

当发布的消息无法路由到任何队列，并且发布者将mandatory消息属性设置为true时，消息将被返回给发布者。发布者必须设置一个返回消息处理程序来处理返回的消息（例如，通过记录错误或尝试使用不同的交换器重新发布）。

备用交换器是AMQP 0-9-1交换器的一项功能，它允许客户端处理交换器无法路由的消息（即，要么是因为没有绑定的队列，要么是没有匹配的绑定）。典型的例子包括检测客户端是否意外或恶意发布无法路由的消息，或者在“否则”路由语义中，一些消息被特殊处理，其余消息由通用处理程序处理。

MQTT

发布到一个新主题会为其设置一个队列。不同的主题/QoS级别组合将使用具有不同属性的不同队列。因此，发布者和消费者必须使用相同的QoS级别。

STOMP

STOMP支持多种不同的目的地，包括那些假设预先存在拓扑的目的地。

  1. /topic/：发布到一个没有消费者的主题将导致消息被丢弃。该主题的第一个订阅者将为其声明一个队列。
  2. /exchange/：目标交换器必须存在，否则服务器将报告错误。
  3. /amq/queue/：目标队列必须存在，否则服务器将报告错误。
  4. /queue/：发布到不存在的队列将创建该队列。
  5. /temp-queue/：发布到不存在的临时队列将创建该队列。

    指标

    有一个关于不可路由丢弃消息的指标： [此处可能是一个图表，展示了一些指标数据，如“Queued messages last minute”（上一分钟排队的消息数）、“Message rates last minute”（上一分钟消息速率）、“Publish”（发布速率）、“Publisher confirm”（发布者确认速率）、“Unroutable (return)”（不可路由返回的消息速率）、“Unroutable (drop)”（不可路由丢弃的消息速率）等]

在上面的示例中，所有发布的消息都因不可路由（且为非强制性）而被丢弃。

消息属性

AMQP 0-9-1

每次消息投递都结合了消息元数据和投递信息。不同的客户端库提供访问这些属性的方式略有不同。通常，投递处理程序可以访问一个投递数据结构。

以下属性是投递和路由详细信息；它们本身不是消息属性，而是由RabbitMQ在路由和投递时设置：

属性类型描述
投递标签（Delivery tag）正整数投递标识符，见确认（Confirms）部分。
重新投递（Redelivered）布尔值如果此消息之前已被投递并重新排队，则设置为true
交换器（Exchange）字符串路由此消息的交换器
路由键（Routing key）字符串发布者使用的路由键
消费者标签（Consumer tag）字符串消费者（订阅）标识符

以下是消息属性。其中大多数是可选的。它们由发布者在发布时设置：

属性类型描述是否必需？
投递模式（Delivery mode）枚举（1或2）2表示“持久化”，1表示“瞬态”。一些客户端库将此属性显示为布尔值或枚举。是
类型（Type）字符串应用程序特定的消息类型，例如“orders.created”否
消息头（Headers）映射（字符串 => 任意类型）一个具有字符串头名称的任意头映射否
内容类型（Content type）字符串内容类型，例如“application/json”。由应用程序使用，不是RabbitMQ核心使用否
内容编码（Content encoding）字符串内容编码，例如“gzip”。由应用程序使用，不是RabbitMQ核心使用否
消息ID（Message ID）字符串任意消息ID否
关联ID（Correlation ID）字符串有助于将请求与响应关联起来，见教程6否
回复地址（Reply To）字符串携带响应队列名称，见教程6否
过期时间（Expiration）字符串每条消息的生存时间（TTL）否
时间戳（Timestamp）时间戳应用程序提供的时间戳否
用户ID（User ID）字符串用户ID，设置时会进行验证否
应用ID（App ID）字符串应用程序名称否

消息类型

消息上的类型属性是一个任意字符串，用于帮助应用程序传达消息的类型。它由发布者在发布时设置。其值可以是发布者和消费者约定的任何特定领域的字符串。

RabbitMQ不会验证或使用此字段，它供应用程序和插件使用和解释。

在实践中，消息类型自然会分为不同的组，使用点分隔的命名约定很常见（但RabbitMQ和客户端并不要求），例如orders.created或logs.line或profiles.image.changed。

如果消费者收到一个未知类型的消息，强烈建议记录此类事件，以便于故障排查。

内容类型和编码

内容（MIME媒体）类型和内容编码字段允许发布者告知消费者应如何反序列化和解码消息有效负载。

RabbitMQ不会验证或使用这些字段，它们供应用程序和插件使用和解释。

例如，带有JSON有效负载的消息应使用application/json。如果有效负载使用LZ77（GZip）算法进行压缩，其内容编码应为gzip。

可以通过用逗号分隔来指定多种编码。

发布者确认（确认机制）和数据安全

确保数据安全是应用程序、客户端库和RabbitMQ集群节点的共同责任。本节涵盖了一些与数据安全相关的主题。

网络可能以不太明显的方式出现故障，检测某些故障需要时间。因此，一个将协议帧或一组帧（例如，一条已发布的消息）写入其套接字的客户端不能假定该消息已到达服务器并被成功处理。消息可能在传输过程中丢失，或者其投递可能会有显著延迟。

为了解决这个问题，开发了一种发布者端确认机制。它模仿了协议中已有的消费者确认机制。

使用发布者确认的策略

发布者确认机制为应用程序开发人员提供了一种跟踪哪些消息已被RabbitMQ成功接受的方法。有几种常用的使用发布者确认的策略：

  1. 单独发布消息并使用流确认（异步API元素：确认事件处理程序、期货/承诺等）。
  2. 批量发布消息并等待所有未完成的确认。
  3. 单独发布消息并等待其被确认后再继续发布。由于这种方式对发布者吞吐量有很大的负面影响，强烈不建议使用。

这些策略在吞吐量影响和易用性方面有所不同。

流确认

大多数客户端库通常为开发人员提供一种处理从服务器到达的单个确认的方法。确认将异步到达。由于在AMQP 0-9-1中发布本质上也是异步的，这种方式允许以很少的开销进行安全发布。其算法通常如下：

  1. 在通道上启用发布者确认。
  2. 对于每一条已发布的消息，添加一个映射项，将当前序列号映射到该消息。
  3. 当收到肯定确认时，删除该项。
  4. 当收到否定确认时，删除该项并安排重新发布该消息（或进行其他合适的操作）。

在RabbitMQ Java客户端中，确认处理程序通过ConfirmCallback和ConfirmListener接口公开。必须向通道添加一个或多个监听器。

批量发布

这种策略涉及批量发布消息并等待整个批次被确认。对批次进行重试。

  1. 在通道上启用发布者确认。
  2. 对于每一批已发布的消息，等待所有未完成的确认。
  3. 当所有确认都为肯定时，发布下一批消息。
  4. 如果有否定确认或超时，重新发布整个批次或仅重新发布相关消息。

一些客户端提供了等待所有未完成确认的便捷API元素。例如，在Java客户端中有Channel#waitForConfirms(timeout)方法。

由于这种方法涉及等待确认，它会对发布者吞吐量产生负面影响。批次越大，影响越小。

发布并等待

这种策略可以被视为一种反模式，记录它主要是为了完整性。它涉及发布一条消息并立即等待未完成的确认到达。可以将其视为上述批量发布策略，其中批次大小等于1。

这种方法会对吞吐量产生非常显著的负面影响，不建议使用。

从连接故障中恢复

客户端和RabbitMQ节点之间的网络连接可能会失败。应用程序如何处理此类故障直接关系到整个系统的数据安全性。

一些RabbitMQ客户端支持自动恢复连接和拓扑结构（队列、交换器、绑定和消费者）：Java、.NET、Bunny等就是一些例子。

其他客户端没有将自动恢复作为一项功能提供，但提供了应用程序开发人员如何实现恢复的示例。

许多应用程序的自动恢复过程遵循以下步骤：

  1. 重新连接到一个可达的节点。
  2. 恢复连接监听器。
  3. 重新打开通道。
  4. 恢复通道监听器。
  5. 恢复通道basic.qos设置、发布者确认和事务设置。

在连接和通道恢复后，可以开始拓扑恢复。拓扑恢复包括以下针对每个通道执行的操作：

  1. 重新声明交换器（预定义的交换器除外）。
  2. 重新声明队列。
  3. 恢复所有绑定。
  4. 恢复所有消费者。

    异常处理

    发布者通常会遇到两种类型的异常：

  5. 由于写入失败或超时导致的网络I/O异常。
  6. 确认投递超时。

请注意，这里的“异常”是指一般意义上的错误；有些编程语言根本没有异常，因此这些语言的客户端会以不同的方式传达错误。本节中的讨论和建议应同样适用于大多数客户端库和编程语言。

前一种类型的异常可能在写入时立即发生，也可能有一定延迟才出现。这是因为某些类型的I/O故障（例如，网络拥塞严重或数据包丢失率高）可能需要一些时间才能检测到。在连接恢复后，发布操作可以继续，但如果连接因警报而被阻塞，在警报解除之前，所有进一步的发布尝试都将失败。在下面“资源警报的影响”部分将对此进行更详细的介绍。

后一种类型的异常只有在应用程序开发人员设置了超时的情况下才会发生。对于特定应用程序而言，合理的超时值由开发人员决定。它不应低于有效的心跳超时时间。

资源警报的影响

当集群节点存在资源警报时，集群中的所有试图发布消息的连接都将被阻塞，直到集群中的所有警报都解除。

当连接被阻塞时，该连接发送的任何数据都不会在连接上被读取、解析或处理。当连接解除阻塞时，客户端的所有流量处理将恢复。

兼容AMQP 0-9-1的客户端在被阻塞和解除阻塞时会收到通知。

在被阻塞的连接上进行写入操作将导致超时或因I/O写入异常而失败。

指标

指标收集和监控对于发布者而言，与应用程序中的任何其他应用程序或组件一样重要。RabbitMQ收集的一些指标与发布者特别相关：

  1. 传出消息速率
  2. 发布者确认速率
  3. 连接变动率
  4. 通道变动率
  5. 不可路由丢弃消息速率
  6. 不可路由返回消息速率

发布速率和确认速率大多不言自明。变动率非常重要，因为它们有助于检测那些没有以最佳方式使用连接或通道的应用程序，这些应用程序的发布速率不理想且会浪费资源。

不可路由消息速率有助于检测那些发布无法路由到任何队列的消息的应用程序。例如，这可能表明存在配置错误。

客户端库也可能收集指标。RabbitMQ Java客户端就是一个例子。这些指标可以深入了解应用程序特定的架构（例如，哪个发布组件发布了不可路由的消息），而这些是RabbitMQ节点无法推断出来的。

并发考虑因素

并发相关的主题主要涉及客户端库的具体实现，但仍可以给出一些通用建议。一般来说，应避免在共享的“发布上下文”（AMQP 0-9-1中的通道、STOMP中的连接、AMQP 1.0中的会话等）上进行发布操作，这种做法被认为是不安全的。

这样做可能会导致网络上的数据帧格式错误，进而导致连接关闭。

在单个应用程序中，如果并发发布者数量较少，为每个发布者使用一个线程（或类似方式）是最佳解决方案。如果并发发布者数量较多（比如数百个或数千个），则可以使用线程池。

临时阻塞发布

通过将内存高水位标记设置为0，可以有效地阻塞集群中的所有发布操作，从而立即触发资源警报：

1

rabbitmqctl set_vm_memory_high_watermark 0

排查发布者问题

本节涵盖了发布者的一些常见问题，以及如何识别和解决这些问题。分布式系统中的故障形式多种多样，因此这个列表并不全面。

连接故障

与任何客户端一样，发布者必须先成功连接并通过身份验证。

潜在的连接问题种类繁多，有专门的指南进行介绍。

身份验证和授权

与任何客户端一样，发布者可能无法通过身份验证，或者没有权限访问其目标虚拟主机，亦或是无法向目标交换器发布消息。

此类故障会被RabbitMQ记录为错误。

有关身份验证和授权问题的排查，请参阅“访问控制指南”中的相关章节。

连接频繁变动

有些应用程序每发布一条消息就打开一个新连接。这效率极低，并且不符合消息传递协议的设计初衷。通过连接指标可以检测到这种情况。

尽可能使用长连接。

连接中断

网络连接可能会中断。一些客户端库支持自动连接和拓扑恢复，而其他客户端库则使得在应用程序代码中实现连接恢复变得容易。

当连接断开时，客户端无法进行发布操作，也不会在内部对消息进行排队（延迟发送）。此外，之前序列化并写入套接字的消息不能保证能到达目标节点。因此，对于需要可靠发布和数据安全的发布者来说，使用发布者确认机制来跟踪哪些发布操作已得到RabbitMQ的确认至关重要。经过一段时间后，未得到确认的消息应被视为未投递成功。如果对应用程序来说是安全的，可以重新发布这些消息。教程7以及本指南中的“数据安全”部分对此进行了介绍。

有关详细信息，请参阅“从网络连接故障中恢复”部分。

路由问题

发布者可能成功连接、通过身份验证并被授予向交换器（主题、目的地）发布消息的权限。然而，这些消息仍有可能无法路由到任何队列或消费者。这可能是由于以下原因：

  1. 应用程序之间的配置不匹配，例如发布者和消费者使用的主题不匹配。
  2. 发布者配置错误（交换器、主题、路由键设置不正确）。
  3. 对于AMQP 0-9-1，目标交换器缺少绑定。
  4. 存在资源警报：见下文。
  5. 网络连接失败且客户端未恢复：见上文。

检查拓扑结构和指标通常有助于快速定位问题。例如，可以使用管理界面中的单个交换器页面来确认是否有入站消息活动（入口速率大于零）以及绑定情况。

在以下示例中，交换器没有绑定，因此没有消息会被路由到任何地方： [此处可能是一个图表，展示了交换器“tmp-fanout”的相关指标数据，如“Message rates last minute”（上一分钟消息速率）、“Publish (In)”（入站发布速率）等，以及“Bindings”（绑定）部分显示“… no bindings …”（没有绑定）]

也可以使用rabbitmq-diagnostics命令列出绑定：

1
2

# 请注意，从RabbitMQ 3.8版本开始，隐式的默认交换器绑定不会被列出
rabbitmq-diagnostics list_bindings --vhost "/"

执行上述命令后，示例中没有返回结果。

从RabbitMQ 3.8版本开始，有一个新的关于不可路由丢弃消息的指标： [此处可能是一个图表，展示了一些指标数据，如“Queued messages last minute”（上一分钟排队的消息数）、“Message rates last minute”（上一分钟消息速率）、“Publish”（发布速率）、“Publisher confirm”（发布者确认速率）、“Unroutable (return)”（不可路由返回的消息速率）、“Unroutable (drop)”（不可路由丢弃的消息速率）等]

在上面的示例中，所有发布的消息都因不可路由（且为非强制性）而被丢弃。有关详细信息，请参阅本指南中的“不可路由消息的处理”部分。

集群范围的指标、连接指标以及服务器日志将有助于发现是否存在资源警报。

资源警报

当存在资源警报时，所有进行发布操作的连接都将被阻塞，直到警报解除。客户端可以选择在被阻塞时接收通知。有关更多信息，请参阅“资源警报指南”。

协议异常

对于某些协议，如AMQP 0-9-1和STOMP，发布者可能会遇到一种称为协议错误（异常）的情况。例如，向不存在的交换器发布消息或把交换器绑定到不存在的交换器上，都会导致通道异常，并使通道关闭。在关闭的通道上无法进行发布操作。这些事件会被发布者所连接的RabbitMQ节点记录下来。根据所使用的客户端库不同，发布失败的尝试还会导致客户端抛出异常或返回错误。

在共享通道上并发发布

客户端库不支持在共享通道上进行并发发布。有关更多信息，请参阅“并发考虑因素”部分。

rabbitmq文档
doc
本文由作者按照 CC BY 4.0 进行授权