Request/Reply Messaging
AmqpTemplate 提供多种 sendAndReceive 方法，用于实现请求-应答模式。这些方法配置必要的消息属性（如相关数据和回复队列），并在内部侦听回复消息。最新版本还支持 direct reply-to，从而消除了固定回复队列的需要。同时，AsyncRabbitTemplate 提供了异步 sendAndReceive 方法，返回 CompletableFuture 以处理请求-应答操作，并支持 direct reply-to。
AmqpTemplate 还提供多种 sendAndReceive 方法，它接受先前为单路发送操作（exchange、routingKey 和 Message）描述的相同参数选项。这些方法对于请求答复场景非常有用，因为它们在发送之前处理必要的 reply-to 特性的配置，并且可以在内部为该目的创建的独占队列中侦听答复消息。
The AmqpTemplate also provides a variety of sendAndReceive methods that accept the same argument options that were described earlier for the one-way send operations (exchange, routingKey, and Message).
Those methods are quite useful for request-reply scenarios, since they handle the configuration of the necessary reply-to property before sending and can listen for the reply message on an exclusive queue that is created internally for that purpose.
同样可用的请求-回复方法是 MessageConverter 同时应用于请求和回复。这些方法命名为 convertSendAndReceive。有关更多详细信息，请参阅 link:https://docs.spring.io/spring-amqp/docs/current/api/org/springframework/amqp/core/AmqpTemplate.html[Javadoc of AmqpTemplate。
Similar request-reply methods are also available where the MessageConverter is applied to both the request and reply.
Those methods are named convertSendAndReceive.
See the Javadoc of AmqpTemplate for more detail.
从版本 1.5.0 开始，每个 sendAndReceive 方法变量都有一个重载版本，其中包含 CorrelationData。与经过适当配置的连接工厂结合使用后，这将启用接收操作发送侧的发布者确认。有关更多信息，请参见 Correlated Publisher Confirms and Returns 和 Javadoc for RabbitOperations。
Starting with version 1.5.0, each of the sendAndReceive method variants has an overloaded version that takes CorrelationData.
Together with a properly configured connection factory, this enables the receipt of publisher confirms for the send side of the operation.
See Correlated Publisher Confirms and Returns and the Javadoc for RabbitOperations for more information.
从版本 2.0 开始，这些方法（convertSendAndReceiveAsType）的变体需要一个附加 ParameterizedTypeReference 参数来转换复杂的返回类型。模板必须使用 SmartMessageConverter 进行配置。有关更多详细信息，请参阅 xref:amqp/message-converters.adoc#json-complex[Converting From a Message With RabbitTemplate。
Starting with version 2.0, there are variants of these methods (convertSendAndReceiveAsType) that take an additional ParameterizedTypeReference argument to convert complex returned types.
The template must be configured with a SmartMessageConverter.
See Converting From a Message With RabbitTemplate for more information.
从 2.1 版开始，您可以使用 noLocalReplyConsumer 选项配置 RabbitTemplate 以控制答复消费者的 noLocal 标志。默认情况下，此值为 false。
Starting with version 2.1, you can configure the RabbitTemplate with the noLocalReplyConsumer option to control a noLocal flag for reply consumers.
This is false by default.
Reply Timeout
默认情况下，发送和接收方法在五秒后超时并返回 null。您可以通过设置 replyTimeout 属性修改此行为。从 1.5 版本开始，如果您将 mandatory 属性设置为 true （或 mandatory-expression 对特定消息求值为 true），则如果无法将消息发送到队列，则会引发 AmqpMessageReturnedException。此异常具有 returnedMessage、replyCode 和 replyText 属性，以及用于发送的 exchange 和 routingKey。
By default, the send and receive methods timeout after five seconds and return null.
You can modify this behavior by setting the replyTimeout property.
Starting with version 1.5, if you set the mandatory property to true (or the mandatory-expression evaluates to true for a particular message), if the message cannot be delivered to a queue, an AmqpMessageReturnedException is thrown.
This exception has returnedMessage, replyCode, and replyText properties, as well as the exchange and routingKey used for the send.
|
此功能使用发布者返回。可以通过将 |
|
This feature uses publisher returns.
You can enable it by setting |
从版本 2.1.2 开始，已添加 replyTimedOut 方法，让子类可以得知超时，以便它们可以清除任何保留的状态。
Starting with version 2.1.2, a replyTimedOut method has been added, letting subclasses be informed of the timeout so that they can clean up any retained state.
从 2.0.11 和 2.1.3 版本开始，当您使用默认 DirectReplyToMessageListenerContainer 时，可以通过设置模板的 replyErrorHandler 属性来添加错误处理程序。对于任何失败的交付（例如：延迟答复和未关联消息头收到的消息），都会调用此错误处理程序。传递的异常是 ListenerExecutionFailedException，它具有 failedMessage 属性。
Starting with versions 2.0.11 and 2.1.3, when you use the default DirectReplyToMessageListenerContainer, you can add an error handler by setting the template’s replyErrorHandler property.
This error handler is invoked for any failed deliveries, such as late replies and messages received without a correlation header.
The exception passed in is a ListenerExecutionFailedException, which has a failedMessage property.
RabbitMQ Direct reply-to
从 3.4.0 版本开始，RabbitMQ 服务器支持 direct reply-to。这消除了固定应答队列的主要原因（为了避免为每个请求创建临时队列）。从 Spring AMQP 1.4.1 版本开始，如果服务器支持，会默认使用直接应答接收（direct reply-to），而不会创建临时应答队列。当未提供 replyQueue（或将名称设置为 amq.rabbitmq.reply-to）时，RabbitTemplate 会自动检测是否支持直接应答接收，进而使用该功能或使用临时应答队列作为后备。当使用直接应答接收时，不需要 reply-listener，也不应进行配置。
Starting with version 3.4.0, the RabbitMQ server supports direct reply-to.
This eliminates the main reason for a fixed reply queue (to avoid the need to create a temporary queue for each request).
Starting with Spring AMQP version 1.4.1 direct reply-to is used by default (if supported by the server) instead of creating temporary reply queues.
When no replyQueue is provided (or it is set with a name of amq.rabbitmq.reply-to), the RabbitTemplate automatically detects whether direct reply-to is supported and either uses it or falls back to using a temporary reply queue.
When using direct reply-to, a reply-listener is not required and should not be configured.
命名队列（而不是 amq.rabbitmq.reply-to）仍然支持答复侦听器，从而实现对答复并发性的控制，等等。
Reply listeners are still supported with named queues (other than amq.rabbitmq.reply-to), allowing control of reply concurrency and so on.
从版本 1.6 开始，如果您希望为每个答复使用临时、独占、自动删除队列，请将 useTemporaryReplyQueues 属性设置为 true。如果您设置 replyAddress，则将忽略此属性。
Starting with version 1.6, if you wish to use a temporary, exclusive, auto-delete queue for each
reply, set the useTemporaryReplyQueues property to true.
This property is ignored if you set a replyAddress.
您可以更改决定是否使用直接答复的条件，通过子类化 RabbitTemplate 并覆盖 useDirectReplyTo() 以检查不同的条件。此方法仅在第一次发送请求时调用一次。
You can change the criteria that dictate whether to use direct reply-to by subclassing RabbitTemplate and overriding useDirectReplyTo() to check different criteria.
The method is called once only, when the first request is sent.
在版本 2.0 之前，RabbitTemplate 为每个请求创建一个新消费者，并在收到答复（或超时）时取消该消费者。现在，模板转而使用 DirectReplyToMessageListenerContainer，让消费者可以重复使用。模板仍然负责关联答复，因此不存在延迟答复转到其他发送者的风险。如果您想要恢复到以前的行为，请将 useDirectReplyToContainer（在使用 XML 配置时为 direct-reply-to-container）属性设置为 false。
Prior to version 2.0, the RabbitTemplate created a new consumer for each request and canceled the consumer when the reply was received (or timed out).
Now the template uses a DirectReplyToMessageListenerContainer instead, letting the consumers be reused.
The template still takes care of correlating the replies, so there is no danger of a late reply going to a different sender.
If you want to revert to the previous behavior, set the useDirectReplyToContainer (direct-reply-to-container when using XML configuration) property to false.
AsyncRabbitTemplate 没有这样的选项。当使用直接答复时，它始终针对答复使用 DirectReplyToContainer。
The AsyncRabbitTemplate has no such option.
It always used a DirectReplyToContainer for replies when direct reply-to is used.
从版本 2.3.7 开始，模板具有一个新属性 useChannelForCorrelation。当此值为 true 时，服务器不必将关联 ID 从请求消息头复制到答复消息。相反，用于发送请求的通道用于关联答复到请求。
Starting with version 2.3.7, the template has a new property useChannelForCorrelation.
When this is true, the server does not have to copy the correlation id from the request message headers to the reply message.
Instead, the channel used to send the request is used to correlate the reply to the request.
Message Correlation With A Reply Queue
在使用固定回复队列（而不是 amq.rabbitmq.reply-to）时，您必须提供相关性数据，以便将回复与请求关联起来。请参见 RabbitMQ Remote Procedure Call (RPC)。默认情况下，使用标准 correlationId 属性来保存相关性数据。但是，如果您希望使用自定义属性来保存相关性数据，则可以在 <rabbit-template/> 上设置 correlation-key 属性。显式将属性设置为 correlationId 与省略属性相同。客户端和服务器必须使用相同的标头来获取相关性数据。
When using a fixed reply queue (other than amq.rabbitmq.reply-to), you must provide correlation data so that replies can be correlated to requests.
See RabbitMQ Remote Procedure Call (RPC).
By default, the standard correlationId property is used to hold the correlation data.
However, if you wish to use a custom property to hold correlation data, you can set the correlation-key attribute on the <rabbit-template/>.
Explicitly setting the attribute to correlationId is the same as omitting the attribute.
The client and server must use the same header for correlation data.
|
Spring AMQP 版本 1.1 为此数据使用了一个名为 |
|
Spring AMQP version 1.1 used a custom property called |
默认情况下，模板生成自己的关联 ID（忽略任何用户提供的数值）。如果您希望使用自己的关联 ID，请将 RabbitTemplate 实例的 userCorrelationId 属性设置为 true。
By default, the template generates its own correlation ID (ignoring any user-supplied value).
If you wish to use your own correlation ID, set the RabbitTemplate instance’s userCorrelationId property to true.
关联 ID 必须唯一，以避免为请求返回错误应答的可能性。
The correlation ID must be unique to avoid the possibility of a wrong reply being returned for a request.
Reply Listener Container
在使用低于 3.4.0 版本的 RabbitMQ 时，会为每个回复使用新的临时队列。 但是，可以在模板上配置一个单独的回复队列，这可以更有效，并且还可以让您在该队列上设置参数。 然而，在这种情况下，您还必须提供一个<reply-listener/>子元素。 该元素为回复队列提供了一个监听器容器，模板是监听器。 <listener-container/>上允许的所有Message Listener Container Configuration属性都允许在该元素上使用，除了`connection-factory`和`message-converter`，它们是从模板的配置中继承的。
When using RabbitMQ versions prior to 3.4.0, a new temporary queue is used for each reply.
However, a single reply queue can be configured on the template, which can be more efficient and also lets you set arguments on that queue.
In this case, however, you must also provide a <reply-listener/> sub element.
This element provides a listener container for the reply queue, with the template being the listener.
All of the Message Listener Container Configuration attributes allowed on a <listener-container/> are allowed on the element, except for connection-factory and message-converter, which are inherited from the template’s configuration.
如果您运行多个应用程序实例或使用多个 RabbitTemplate 实例，则 MUST 为每个应用程序实例使用一个唯一的应答队列。RabbitMQ 无法从队列中选择消息，因此，如果它们都使用相同的队列，则每个实例都会争夺应答，而不一定能够收到它们自己的应答。
If you run multiple instances of your application or use multiple RabbitTemplate instances, you MUST use a unique reply queue for each.
RabbitMQ has no ability to select messages from a queue, so, if they all use the same queue, each instance would compete for replies and not necessarily receive their own.
以下示例定义了一个带有连接工厂的 Rabbit 模板：
The following example defines a rabbit template with a connection factory:
<rabbit:template id="amqpTemplate"
connection-factory="connectionFactory"
reply-queue="replies"
reply-address="replyEx/routeReply">
<rabbit:reply-listener/>
</rabbit:template>虽然容器和模板共享一个连接工厂，但它们并不共享通道。因此，请求和答复不在同一事务中执行（如果它具有事务性）。
While the container and template share a connection factory, they do not share a channel. Therefore, requests and replies are not performed within the same transaction (if transactional).
|
在版本 1.5.0 之前， |
|
Prior to version 1.5.0, the |
通过此配置，SimpleListenerContainer 用于接收答复，其中 RabbitTemplate 是 MessageListener。在模板中定义一个带有 <rabbit:template/> 命名空间元素时，如前一个示例所示，解析器将定义容器并将模板作为侦听器连接起来。
With this configuration, a SimpleListenerContainer is used to receive the replies, with the RabbitTemplate being the MessageListener.
When defining a template with the <rabbit:template/> namespace element, as shown in the preceding example, the parser defines the container and wires in the template as the listener.
|
当模板不使用固定 |
|
When the template does not use a fixed |
如果您将 RabbitTemplate 定义为 <bean/> 或使用 @Configuration 类来将其定义为 @Bean 或以编程方式创建模板，则需要自行定义和连接回复侦听器容器。如果您未执行此操作，则模板永远不会收到答复，最终超时并将 null 返回为 sendAndReceive 方法调用的答复。
If you define your RabbitTemplate as a <bean/> or use an @Configuration class to define it as an @Bean or when you create the template programmatically, you need to define and wire up the reply listener container yourself.
If you fail to do this, the template never receives the replies and eventually times out and returns null as the reply to a call to a sendAndReceive method.
从 1.5 版开始，RabbitTemplate 检测是否已将它配置为 MessageListener 以接收答复。如果否，则尝试使用答复地址发送和接收消息会导致 IllegalStateException（因为永远不会收到答复）。
Starting with version 1.5, the RabbitTemplate detects if it has been
configured as a MessageListener to receive replies.
If not, attempts to send and receive messages with a reply address
fail with an IllegalStateException (because the replies are never received).
此外，如果使用了简单的 replyAddress（队列名称），则回复侦听器容器将验证它是否正在侦听具有相同名称的队列。如果答复地址是交换机和路由密钥，则无法执行此检查，并且会写入调试日志消息。
Further, if a simple replyAddress (queue name) is used, the reply listener container verifies that it is listening
to a queue with the same name.
This check cannot be performed if the reply address is an exchange and routing key and a debug log message is written.
在自行连接回复侦听器和模板时，请务必确保模板的 replyAddress 和容器的 queues（或 queueNames）属性引用同一队列，模板将回复地址插入出站消息 replyTo 属性中。
When wiring the reply listener and template yourself, it is important to ensure that the template’s replyAddress and the container’s queues (or queueNames) properties refer to the same queue.
The template inserts the reply address into the outbound message replyTo property.
以下清单显示了如何手动连接 bean 的示例：
The following listing shows examples of how to manually wire up the beans:
<bean id="amqpTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate">
<constructor-arg ref="connectionFactory" />
<property name="exchange" value="foo.exchange" />
<property name="routingKey" value="foo" />
<property name="replyQueue" ref="replyQ" />
<property name="replyTimeout" value="600000" />
<property name="useDirectReplyToContainer" value="false" />
</bean>
<bean class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
<constructor-arg ref="connectionFactory" />
<property name="queues" ref="replyQ" />
<property name="messageListener" ref="amqpTemplate" />
</bean>
<rabbit:queue id="replyQ" name="my.reply.queue" /> @Bean
public RabbitTemplate amqpTemplate() {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
rabbitTemplate.setMessageConverter(msgConv());
rabbitTemplate.setReplyAddress(replyQueue().getName());
rabbitTemplate.setReplyTimeout(60000);
rabbitTemplate.setUseDirectReplyToContainer(false);
return rabbitTemplate;
}
@Bean
public SimpleMessageListenerContainer replyListenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueues(replyQueue());
container.setMessageListener(amqpTemplate());
return container;
}
@Bean
public Queue replyQueue() {
return new Queue("my.reply.queue");
}一个使用固定回复队列连接的 RabbitTemplate 的完整示例，以及处理请求并返回回复的 “remote” 侦听器容器显示在 this test case 中。
A complete example of a RabbitTemplate wired with a fixed reply queue, together with a “remote” listener container that handles the request and returns the reply is shown in this test case.
当回复超时（replyTimeout）时，sendAndReceive() 方法返回 null。
When the reply times out (replyTimeout), the sendAndReceive() methods return null.
在 1.3.6 版之前，超时的消息的延迟答复只会记录下来。现在，如果收到延迟答复，则会将其拒绝（模板会抛出 AmqpRejectAndDontRequeueException）。如果回复队列配置为将被拒绝的消息发送到死信交换机，则可以检索答复以进行后续分析。为此，请使用与回复队列名称相等的路由密钥将队列绑定到配置的死信交换机。
Prior to version 1.3.6, late replies for timed out messages were only logged.
Now, if a late reply is received, it is rejected (the template throws an AmqpRejectAndDontRequeueException).
If the reply queue is configured to send rejected messages to a dead letter exchange, the reply can be retrieved for later analysis.
To do so, bind a queue to the configured dead letter exchange with a routing key equal to the reply queue’s name.
有关配置死信的更多信息，请参见 RabbitMQ Dead Letter Documentation。您还可以查看 FixedReplyQueueDeadLetterTests 测试用例作为示例。
See the RabbitMQ Dead Letter Documentation for more information about configuring dead lettering.
You can also take a look at the FixedReplyQueueDeadLetterTests test case for an example.
Async Rabbit Template
版本 1.6 引入了`AsyncRabbitTemplate`。 这具有与 AmqpTemplate上类似的`sendAndReceive`（和`convertSendAndReceive`）方法。 但是，它们不阻塞，而是返回`CompletableFuture`。
Version 1.6 introduced the AsyncRabbitTemplate.
This has similar sendAndReceive (and convertSendAndReceive) methods to those on the AmqpTemplate.
However, instead of blocking, they return a CompletableFuture.
sendAndReceive 方法返回 RabbitMessageFuture。convertSendAndReceive 方法返回 RabbitConverterFuture。
The sendAndReceive methods return a RabbitMessageFuture.
The convertSendAndReceive methods return a RabbitConverterFuture.
您可以通过在 future 上调用 get() 来同步检索结果，或者您可以注册一个回调，该回调将异步地使用结果调用。以下清单显示了两种方法：
You can either synchronously retrieve the result later, by invoking get() on the future, or you can register a callback that is called asynchronously with the result.
The following listing shows both approaches:
@Autowired
private AsyncRabbitTemplate template;
...
public void doSomeWorkAndGetResultLater() {
...
CompletableFuture<String> future = this.template.convertSendAndReceive("foo");
// do some more work
String reply = null;
try {
reply = future.get(10, TimeUnit.SECONDS);
}
catch (ExecutionException e) {
...
}
...
}
public void doSomeWorkAndGetResultAsync() {
...
RabbitConverterFuture<String> future = this.template.convertSendAndReceive("foo");
future.whenComplete((result, ex) -> {
if (ex == null) {
// success
}
else {
// failure
}
});
...
}如果设置了 mandatory 且无法传递消息，则 future 会抛出 ExecutionException，其原因是 AmqpMessageReturnedException，它封装了返回的消息和有关返回的信息。
If mandatory is set and the message cannot be delivered, the future throws an ExecutionException with a cause of AmqpMessageReturnedException, which encapsulates the returned message and information about the return.
如果设置了 enableConfirms，则 future 具有一个名为 confirm 的属性，它本身是一个 CompletableFuture<Boolean>，其中 true 表示发布成功。如果确认 future 为 false，则 RabbitFuture 有一个名为 nackCause 的进一步属性，其中包含失败原因（如果可用）。
If enableConfirms is set, the future has a property called confirm, which is itself a CompletableFuture<Boolean> with true indicating a successful publish.
If the confirm future is false, the RabbitFuture has a further property called nackCause, which contains the reason for the failure, if available.
如果在回复之后收到发布确认，则将其丢弃，因为回复表示成功发布。
The publisher confirm is discarded if it is received after the reply, since the reply implies a successful publish.
您可以设置模板上的 receiveTimeout 属性以使回复超时（它默认为 30000 - 30 秒）。如果发生超时，则 future 将完成 AmqpReplyTimeoutException。
You can set the receiveTimeout property on the template to time out replies (it defaults to 30000 - 30 seconds).
If a timeout occurs, the future is completed with an AmqpReplyTimeoutException.
此模板实现 SmartLifecycle。在有待处理的答复时停止模板会导致待处理的 Future 实例被取消。
The template implements SmartLifecycle.
Stopping the template while there are pending replies causes the pending Future instances to be canceled.
从版本 2.0 开始，异步模板现在支持 direct reply-to，而不是配置的回复队列。要启用此功能，请使用以下构造函数之一：
Starting with version 2.0, the asynchronous template now supports direct reply-to instead of a configured reply queue. To enable this feature, use one of the following constructors:
public AsyncRabbitTemplate(ConnectionFactory connectionFactory, String exchange, String routingKey)
public AsyncRabbitTemplate(RabbitTemplate template)请参见 RabbitMQ Direct reply-to，直接使用回复到与同步 RabbitTemplate 一起使用。
See RabbitMQ Direct reply-to to use direct reply-to with the synchronous RabbitTemplate.
2.0 版引入了这些方法变体（convertSendAndReceiveAsType），该变体需要一个附加 ParameterizedTypeReference 参数来转换复杂的返回类型。你必须使用 SmartMessageConverter 对底层 RabbitTemplate 进行配置。有关更多详细信息，请参阅 xref:amqp/message-converters.adoc#json-complex[Converting From a Message With RabbitTemplate。
Version 2.0 introduced variants of these methods (convertSendAndReceiveAsType) that take an additional ParameterizedTypeReference argument to convert complex returned types.
You must configure the underlying RabbitTemplate with a SmartMessageConverter.
See Converting From a Message With RabbitTemplate for more information.
从版本 3.0 开始，AsyncRabbitTemplate 方法现在返回 CompletableFuture 而不是 ListenableFuture。
Starting with version 3.0, the AsyncRabbitTemplate methods now return CompletableFuture s instead of ListenableFuture s.
Spring Remoting with AMQP
不再支持 Spring remoting，因为该功能已从 Spring 框架中移除。
Spring remoting is no longer supported because the functionality has been removed from Spring Framework.
改为使用 RabbitTemplate（客户端）和 @RabbitListener 上的 sendAndReceive 操作。
Use sendAndReceive operations using the RabbitTemplate (client side ) and @RabbitListener instead.