跳到主要内容
版本：4.1.1

发送消息

Doubao 中英对照 Sending Messages

发送消息时，你可以使用以下任一方法：

void send(Message message) throws AmqpException;

void send(String routingKey, Message message) throws AmqpException;

void send(String exchange, String routingKey, Message message) throws AmqpException;

我们可以从前面清单中的最后一个方法开始讨论，因为它实际上是最显式的。它允许在运行时提供 AMQP 交换机名称（以及路由键）。最后一个参数是负责实际创建消息实例的回调。使用此方法发送消息的例子可能如下所示：下例展示了如何使用 send 方法发送消息：

amqpTemplate.send("marketData.topic", "quotes.nasdaq.THING1",
    new Message("12.34".getBytes(), someProperties));

如果你打算在大多数或所有情况下都使用该模板实例向同一个交换机发送消息，可以在模板本身上设置 exchange 属性。在这种情况下，你可以使用前面清单中的第二个方法。下例在功能上等同于上例：

amqpTemplate.setExchange("marketData.topic");
amqpTemplate.send("quotes.nasdaq.FOO", new Message("12.34".getBytes(), someProperties));

如果模板上同时设置了 exchange 和 routingKey 属性，你可以使用只接收 Message 的方法。下例展示了如何这样做：

amqpTemplate.setExchange("marketData.topic");
amqpTemplate.setRoutingKey("quotes.nasdaq.FOO");
amqpTemplate.send(new Message("12.34".getBytes(), someProperties));

理解交换机和路由键属性的更好方式是：显式的方法参数总是会覆盖模板的默认值。事实上，即使你没有在模板上显式设置这些属性，也始终有默认值存在。在这两种情况下，默认值都是空 String，但这实际上是一个合理的默认值。就路由键而言，它本来就并非总是必需的（例如对于 Fanout 交换机）。此外，队列也可以以空 String 绑定到交换机。这两种情况都可以依赖模板路由键属性默认的空 String 值。就交换机名称而言，空 String 很常用，因为 AMQP 规范将“默认交换机”定义为没有名称。由于所有队列都会自动绑定到那个默认交换机（它是一个 direct 交换机），并以其名称作为绑定值，前面清单中的第二个方法可用于通过默认交换机向任何队列进行简单的点对点消息传递。你可以在运行时提供方法参数，将队列名作为 routingKey 传入。下例展示了如何这样做：

RabbitTemplate template = new RabbitTemplate(); // using default no-name Exchange
template.send("queue.helloWorld", new Message("Hello World".getBytes(), someProperties));

或者，你可以创建一个主要或专门用于向单个队列发布消息的模板。下例展示了如何这样做：

RabbitTemplate template = new RabbitTemplate(); // using default no-name Exchange
template.setRoutingKey("queue.helloWorld"); // but we'll always send to this Queue
template.send(new Message("Hello World".getBytes(), someProperties));

消息构建器 API​

从 1.3 版本开始，MessageBuilder 和 MessagePropertiesBuilder 提供了消息构建器 API。这些方法提供了一种便捷的“流式”方式来创建消息或消息属性。以下示例展示了流式 API 的实际用法：

Message message = MessageBuilder.withBody("foo".getBytes())
    .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
    .setMessageId("123")
    .setHeader("bar", "baz")
    .build();
MessageProperties props = MessagePropertiesBuilder.newInstance()
    .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
    .setMessageId("123")
    .setHeader("bar", "baz")
    .build();
Message message = MessageBuilder.withBody("foo".getBytes())
    .andProperties(props)
    .build();

MessageProperties 上定义的每个属性都可以设置。其他方法包括 setHeader(String key, String value)、removeHeader(String key)、removeHeaders() 和 copyProperties(MessageProperties properties)。每个属性设置方法都有一个 set*IfAbsent() 变体。在存在默认初始值的情况下，方法名为 set*IfAbsentOrDefault()。

提供了五个静态方法来创建初始消息构建器：

public static MessageBuilder withBody(byte[] body) 1

public static MessageBuilder withClonedBody(byte[] body) 2

public static MessageBuilder withBody(byte[] body, int from, int to) 3

public static MessageBuilder fromMessage(Message message) 4

public static MessageBuilder fromClonedMessage(Message message) 5

  • 构建器创建的消息，其消息体是对参数的直接引用。

  • 构建器创建的消息，其消息体是一个新数组，包含参数字节的副本。

  • 构建器创建的消息，其消息体是一个新数组，包含参数中指定范围内的字节。更多细节参见 Arrays.copyOfRange()。

  • 构建器创建的消息，其消息体是对参数消息体的直接引用。参数的属性被复制到一个新的 MessageProperties 对象。

  • 构建器创建的消息，其消息体是一个新数组，包含参数消息体的副本。参数的属性被复制到一个新的 MessageProperties 对象。

提供了三个静态方法来创建 MessagePropertiesBuilder 实例：

public static MessagePropertiesBuilder newInstance() 1

public static MessagePropertiesBuilder fromProperties(MessageProperties properties) 2

public static MessagePropertiesBuilder fromClonedProperties(MessageProperties properties) 3

  • 新的消息属性对象以默认值初始化。

  • 构建器以提供的属性对象初始化，且 build() 将返回该对象。

  • 参数的属性被复制到一个新的 MessageProperties 对象。

对于 AmqpTemplate 的 RabbitTemplate 实现，每个 send() 方法都有一个重载版本，它额外接收一个 CorrelationData 对象。当启用发布确认时，该对象会在 AmqpTemplate 中所述的回调中返回。这让发送方可以将确认（ack 或 nack）与已发送的消息关联起来。

从 1.6.7 版本开始，引入了 CorrelationAwareMessagePostProcessor 接口，允许在消息转换之后修改关联数据。下例展示了如何使用它：

Message postProcessMessage(Message message, Correlation correlation);

在 2.0 版本中，此接口已被弃用。该方法已被移至 MessagePostProcessor，并带有一个委托给 postProcessMessage(Message message) 的默认实现。

同样从 1.6.7 版本开始，提供了一个名为 CorrelationDataPostProcessor 的新回调接口。它在所有 MessagePostProcessor 实例（send() 方法中提供的以及 setBeforePublishPostProcessors() 中提供的）之后被调用。实现可以更新或替换 send() 方法中提供的关联数据（如果有）。Message 和原始 CorrelationData（如果有）作为参数提供。下例展示了如何使用 postProcess 方法：

CorrelationData postProcess(Message message, CorrelationData correlationData);

发布者退回​

当模板的 mandatory 属性为 true 时，退回的消息会通过 AmqpTemplate 中所述的回调提供。

从 1.4 版本开始，RabbitTemplate 支持 SpEL mandatoryExpression 属性，它以每条请求消息作为根求值对象进行求值，解析为一个 boolean 值。表达式中可以使用 bean 引用，例如 @myBean.isMandatory(#root)。

发布者退回也可以在 RabbitTemplate 的发送和接收操作中内部使用。更多信息参见回复超时。

批量处理​

1.4.2 版本引入了 BatchingRabbitTemplate。它是 RabbitTemplate 的子类，重写了 send 方法，根据 BatchingStrategy 对消息进行批量处理。只有当批次完成时，消息才会发送到 RabbitMQ。下面的清单展示了 BatchingStrategy 接口的定义：

public interface BatchingStrategy {

    MessageBatch addToBatch(String exchange, String routingKey, Message message);

    Date nextRelease();

    Collection<MessageBatch> releaseBatches();

}
警告

批量数据保存在内存中。系统发生故障时，未发送的消息可能会丢失。

提供了一个 SimpleBatchingStrategy。它支持向单个交换机或路由键发送消息。它具有以下属性：

  • batchSize：批次在发送前包含的消息条数。
  • bufferLimit：批量消息的最大大小。如果超过此值，它会优先于 batchSize，并导致部分批次被发送。
  • timeout：当没有新活动向批次添加消息时，经过该时间后会发送部分批次。

SimpleBatchingStrategy 通过在每条内嵌消息前添加一个四字节二进制长度来格式化批次。这通过将 springBatchFormat 消息属性设置为 lengthHeader4 来告知接收系统。

important

默认情况下，监听器容器会自动对批量消息进行拆批（通过使用 springBatchFormat 消息头）。拒绝批次中的任何一条消息都会导致整个批次被拒绝。

不过，更多信息请参见带批量处理的 @RabbitListener。