RabbitMQ AMQP.BasicProperties.Builder值

48

在RabbitMQ/AMQP的Java客户端中,您可以创建一个 AMQP.BasicProperties.Builder,并使用它来build()一个 AMQP.BasicProperties 实例。这个建立的属性实例可以用于各种重要的事情。在这个构建器类上有许多“构建器”样式的方法可用:

BasicProperties.Builder propsBuilder = new BasicProperties.Builder();
propsBuilder
    .appId(???)
    .clusterId(???)
    .contentEncoding(???)
    .contentType(???)
    .correlationId(???)
    .deliveryMode(2)
    .expiration(???)
    .headers(???)
    .messageId(???)
    .priority(???)
    .replyTo(???)
    .timestamp(???)
    .type(???)
    .userId(???);
我正在寻找这些Builder方法帮助“构建”的字段,最重要的是,每个字段存在哪些有效值。例如,什么是clusterId,它的有效值是什么?什么是type,它的有效值是什么?等等。
我已经花了整个上午搜寻: 在所有这些文档中,我都找不到清晰的定义(除了对prioritycontentEncodingdeliveryMode的一些模糊解释),说明每个字段是什么以及它们的有效值是什么。有人知道吗?更重要的是,有人知道这些甚至在哪里记录吗?提前致谢!

1
有一天,为Java编写这些API的人们将会了解ENUMS。 - PedroD
3个回答

89
通常我使用非常简单的方法来记忆某些内容。我将在下面提供所有细节,但这里有一个基本属性字段和值的简单图片。我还试图正确突出队列/服务器和应用程序上下文。

enter image description here

If you want me to enhance it a bit - just drop a small comment. What I really want is to provide some visual key and simplify understanding.
高级描述(来源1来源2):
请注意,Clust ID已被弃用,因此我将排除它。
  • 应用程序ID - 生成消息的应用程序的标识符。
    • 上下文:应用程序使用
    • 值:可以是任何字符串。
  • 内容编码 - 消息内容编码。
    • 上下文:应用程序使用
    • 值:MIME内容编码(例如gzip)
  • 内容类型 - 消息内容类型。
    • 上下文:应用程序使用
    • 值:MIME内容类型(例如application/json)
  • 相关ID - 与此消息相关的消息,例如这条消息回复了哪个请求。鼓励应用程序使用此属性,而不是将此信息放入消息有效负载中。
    • 上下文:应用程序使用
    • 值:任何值
  • 传递模式 - 是否将消息持久化到磁盘?
    • 上下文:队列实现使用
    • 值:非持久性(1)或持久性(2)
  • 过期时间 - 消息将被删除的过期时间。过期字段的值描述以毫秒为单位的TTL周期。请参见下面的详细信息。
    • 上下文:队列实现使用
  • 标头 - 任意应用程序特定的消息标头。
    • 上下文:应用程序使用
  • 消息ID - 消息标识符字符串。如果应用程序需要标识消息,则建议使用此属性,而不是将其放入消息有效负载中。
    • 上下文:应用程序使用
    • 值:任何值
  • 优先级 - 消息优先级。
    • 上下文:队列实现使用
    • 值:0至9
  • 回复地址 - 其他应用程序应将响应发送到的队列名称。通常用于命名回复队列(或任何其他标识符,可帮助消费者应用程序指示其响应)。鼓励应用程序使用此属性,而不是将此信息放入消息有效负载中。
    • 上下文:应用程序使用
    • 值:任何值
  • 时间戳 - 发送消息时的时间戳。
    • 上下文:应用程序使用
    • 值:自纪元以来的秒数。
  • 类型 - 消息类型,例如此消息表示的事件或命令类型。建议应用程序使用此属性,而不是将此信息包含在消息有效负载中。
    • 上下文:应用程序使用
    • 值:可以是任何字符串。
  • 用户ID - 可选的用户ID。由RabbitMQ针对实际连接用户名进行验证。
    • 上下文:队列实现使用
    • 值:应该是经过身份验证的用户。

顺便说一下,我终于成功地审查了最新的服务器代码(rabbitmq-server-3.1.5),rabbit_stomp_test_util.erl中有一个例子:

                content_type     = <<"text/plain">>,
                content_encoding = <<"UTF-8">>,
                delivery_mode    = 2,
                priority         = 1,
                correlation_id   = <<"123">>,
                reply_to         = <<"something">>,
                expiration       = <<"my-expiration">>,
                message_id       = <<"M123">>,
                timestamp        = 123456,
                type             = <<"freshly-squeezed">>,
                user_id          = <<"joe">>,
                app_id           = <<"joe's app">>,
                headers          = [{<<"str">>, longstr, <<"foo">>},
                                    {<<"int">>, longstr, <<"123">>}]

很高兴有人想知道所有细节。因为在可能的情况下,使用众所周知的消息属性比将信息放在消息正文中要好得多。顺便说一句,基本的消息属性远非清晰和有用。我会说最好使用自定义属性。

enter image description here

好的例子 (source)

enter image description here

更新 - 过期时间字段

重要提示: 过期时间 属于队列上下文。因此,消息可能会被服务器丢弃。

enter image description here

README中写道:

expiration是一个shortstr类型的变量;RabbitMQ希望这是一个编码过的字符串,所以我们将ttl翻译为其整数值的字符串表示。

参考资料:


Content Encoding 和 Content Type 的值描述需要交换。 - Slaven Rezic
@SlavenRezic,你说得非常正确。对于我的错误我很抱歉,谢谢你。 - Renat Gilmanov
非常好的答案,谢谢。不过,“附加来源2”的链接已经失效了。 - Max Barraclough
在时间戳字段中,类型实际上是时间戳而不是数字(包含毫秒),Rabbit管理页面将其显示为数字,但如果您正在接收消息,则可以看到它,并且声明也告诉您他期望一个时间戳。 - Thomas

10

撰写时:

  1. 最新的AMQP标准是AMQP 1.0 OASIS Standard
  2. RabbitMQ的最新版本为3.1.5(服务器和客户端),声称支持AMQP 0.9.1(pdf和XML模式压缩)。
  3. RabbitMQ提供了自己的协议描述,包括扩展XML模式(即非标准模式),以及不带扩展的XML模式(与通过(2)链接的模式相同)和pdf文档

在这个答案中:

  • 第三项链接是详细信息的主要来源。
  • 如果第三项不足够详细,第二项pdf文档可用作辅助材料。
  • 如果第二项也不足够详细,则使用源代码(java客户端、erlang服务器)作为第三级详细信息。
  • 通常不使用第一项 - 协议和模式已经在OASIS中进行了(相当)重大的演变,并将适用于未来版本的RabbitMQ,但现在不适用。只有两个例外情况使用了第一项 - 用于contentTypecontentEncoding的文本描述 - 这是安全的,因为这些是具有良好描述的标准字段在AMQP 1.0中。

以下文本是我根据这些来源进行的改编,使其更加简明或清晰。

  • content-type (AMQP XML type="shortstr"; java type="String"): 可选。消息应用数据部分(正文)的 RFC-2046 MIME 类型。可以包含一个字符集参数,定义使用的字符编码,例如"text/plain; charset=“utf-8”"。如果内容类型未知,则不应设置 content-type,以允许接收方确定实际类型。如果该部分已知为真正的不透明二进制数据,则应将 content-type 设置为application/octet-stream。
  • content-encoding (AMQP XML type="shortstr"; java type="String"): 可选。存在时,描述应用数据应用的附加内容编码,因此需要应用哪些解码机制才能获得由 content-type 标头字段引用的媒体类型。主要用于允许压缩文档而不丢失其基础内容类型的标识。作为 content-type 的修饰符,根据 RFC 2616 第 3.5 节进行解释。有效的内容编码在 IANA 注册。实现不应使用 compress 编码,除非与最初使用其他协议(例如 HTTP 或 SMTP)发送的消息保持兼容。实现不应指定多个 content-encoding 值,除非与最初使用其他协议(例如 HTTP 或 SMTP)发送的消息保持兼容。
  • headers (AMQP XML type="table"; java type="Map"): 可选。应用程序指定的头参数及其值的列表。可以仅为应用程序使用而设置。此外,还可以创建具有“Header Exchange Type”的队列 - 当创建队列时,它会获得一系列要匹配的头属性名称,每个名称都有要匹配的可选值,以便通过头匹配路由到该队列。
  • deliveryMode (RabbitMQ XML type="octet"; java type="Integer"): 1(非持久性)或2(持久性)。仅适用于实现持久性的队列。持久性消息安全地存储在磁盘上,并保证在发生严重的网络故障、服务器崩溃、溢出等情况下能够传递。
  • priority (AMQP XML type="octet"; java type="Integer"): 相对的消息优先级(0 到 9)。高优先级消息[MAY BE?? - GB]比同一消息队列中等待的低优先级消息优先发送。当必须丢弃消息以维护特定的服务质量水平时,服务器将首先丢弃低优先级消息。仅适用于实现优先级的队列。
  • correlation-id (AMQP XML type="octet"; java type="String"): 可选。供应用程序使用,没有正式的(RabbitMQ)行为。客户端特定的 ID,可用于标记或标识客户端之间的消息。
  • replyTo (AMQP XML type="shortstr"; java type="String"): 可选。供应用程序使用,没有正式的(RabbitMQ)行为,但可能包含一个私有响应队列的名称,在请求消息中使用时。发送回复的节点的地址。
  • expiration (AMQP XML type="shortstr"; java type="String"): 可选。RabbitMQ AMQP 0.9.1 schema from (3) states "For implementation use, no formal behaviour"。来自(2)的 AMQP 0.9.1 模式 pdf 指出了此消息被视为已过期的绝对时间。但是,必须忽略{{link2:这个
    正如您在上面看到的,绝大多数这些属性都没有枚举/约束/推荐值,因为它们仅用于应用程序,不被RabbitMQ使用。所以你的工作很简单。只要匹配数据类型和编译,就可以自由地编写/读取对您的应用程序有用的值 :). ContentType和contentEncoding按照标准HTTP使用。DeliveryMode和priority是受限数字。
    注意:AMQP.BasicProperties的有用但简单的常量可在类MessageProperties中获得。
    干杯 :)
    更新帖子:
    非常感谢Renat(请参见评论),已查看rabbit_amqqueue_process.erl中的erlang服务器代码和RabbitMQ TTL Extensions to AMQP文档。可以指定消息过期时间(生存时间)。
    • per queue via:

      Map<String, Object> args = new HashMap<String, Object>();
      args.put("x-message-ttl", 60000);
      channel.queueDeclare("myqueue", false, false, false, args);
      
    • or per message via:

      byte[] messageBodyBytes = "Hello, world!".getBytes();
      AMQP.BasicProperties properties = new AMQP.BasicProperties();
      properties.setExpiration("60000");
      channel.basicPublish("my-exchange", "routing-key", properties, messageBodyBytes);
      
    在这里,ttl/过期时间以毫秒为单位,因此每种情况下为60秒。 已更新上述过期定义以反映此内容。

@Glen_Best,过期时间不适用于应用程序。请检查rabbit_amqqueue_process.erl或查看我的更新。 - Renat Gilmanov
已经检查了代码 - 你是正确的。已编辑帖子。谢谢! - Glen Best
1
澄清:根据 https://www.rabbitmq.com/ttl.html,此为应用程序使用;我的解释并未表明 Rabbit 进行正式处理 - 事实上是有这样的流程的。即 RabbitMQ 0.9.1 模式注释已经过时 / 错误。 - Glen Best

0

AMQP规范定义了一种通用的、可扩展的属性模型。

AMQP属性在概念上与HTTP头类似,因为它们表示有关消息的元数据。就像在HTTP中一样,它们与消息有效载荷分开。但它们基本上是一个键/值映射。

一些代理(如RabbitMQ)会解释某些消息属性,例如expiration,以添加额外的供应商特定值(在这种情况下,强制执行TTL)。

但最终,AMQP属性只是一大堆键/值对,如果您选择这样做,它们将安全地随每个消息一起发送。您的AMQP代理的文档将告诉您哪些属性他们特别解释以及如何发送自己的属性。

话虽如此,如果您首先提出这个问题,那么您可能根本不需要担心它们。您将能够成功发送消息,而无需担心设置任何消息属性。


网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接