跳到主要内容
版本:4.1.1

请求/回复消息传递

Doubao 中英对照 Request/Reply Messaging

AmqpTemplate 还提供了多种 sendAndReceive 方法,它们接受与前面所述单向发送操作相同的参数选项(exchange、routingKey 和 Message)。这些方法对于请求-回复场景非常有用,因为它们在发送前处理必要的 reply-to 属性配置,并可以在内部为此目的创建的独占队列上监听回复消息。

类似的请求-回复方法也有同时对请求和回复应用 MessageConverter 的版本。这些方法名为 convertSendAndReceive。更多细节参见 AmqpTemplate 的 Javadoc。

从 1.5.0 版本开始,每个 sendAndReceive 方法变体都有一个接收 CorrelationData 的重载版本。配合正确配置的连接工厂,这可以在操作的发送一侧接收发布确认。更多信息参见关联的发布确认与退回和 RabbitOperations 的 Javadoc。

从 2.0 版本开始,这些方法还有一些变体(convertSendAndReceiveAsType),它们额外接收一个 ParameterizedTypeReference 参数,用于转换复杂的返回类型。模板必须配置一个 SmartMessageConverter。更多信息参见用 RabbitTemplate 从消息转换。

从 2.1 版本开始,你可以用 noLocalReplyConsumer 选项配置 RabbitTemplate,以控制回复消费者的 noLocal 标志。默认为 false。

回复超时​

默认情况下,发送和接收方法在五秒后超时并返回 null。你可以通过设置 replyTimeout 属性来修改此行为。从 1.5 版本开始,如果你将 mandatory 属性设置为 true(或 mandatory-expression 对某条消息求值为 true),如果消息无法投递到队列,就会抛出 AmqpMessageReturnedException。此异常带有 returnedMessage、replyCode 和 replyText 属性,以及发送时使用的 exchange 和 routingKey。

备注

此功能使用发布者退回。你可以通过在 CachingConnectionFactory 上把 publisherReturns 设置为 true 来启用它(参见发布确认与退回)。此外,你不能在 RabbitTemplate 上注册自己的 ReturnCallback。

从 2.1.2 版本开始,新增了一个 replyTimedOut 方法,让子类可以得知超时,从而清理任何保留的状态。

从 2.0.11 和 2.1.3 版本开始,当你使用默认的 DirectReplyToMessageListenerContainer 时,可以通过设置模板的 replyErrorHandler 属性来添加一个错误处理器。对于任何失败投递(例如迟到的回复以及没有关联头而收到的消息),都会调用此错误处理器。传入的异常是 ListenerExecutionFailedException,它带有一个 failedMessage 属性。

RabbitMQ 直接回复(direct reply-to)​

important

从 3.4.0 版本开始,RabbitMQ 服务器支持直接回复。这消除了使用固定回复队列的主要原因(避免为每个请求创建临时队列)。从 Spring AMQP 1.4.1 版本开始,默认使用直接回复(如果服务器支持),而不是创建临时回复队列。当未提供 replyQueue(或将其名称设为 amq.rabbitmq.reply-to)时,RabbitTemplate 会自动检测是否支持直接回复,并据此选择使用它或回退到临时回复队列。使用直接回复时,不需要 reply-listener,也不应配置它。

回复监听器仍支持命名队列(amq.rabbitmq.reply-to 除外),从而可以控制回复并发等。

从 1.6 版本开始,如果你希望为每次回复使用一个临时、独占、自动删除的队列,请将 useTemporaryReplyQueues 属性设置为 true。如果你设置了 replyAddress,此属性会被忽略。

你可以通过子类化 RabbitTemplate 并重写 useDirectReplyTo() 来检查不同的条件,从而更改是否使用直接回复的判定标准。该方法仅在第一个请求发送时调用一次。

在 2.0 版本之前,RabbitTemplate 会为每个请求创建一个新消费者,并在收到回复(或超时)时取消该消费者。现在模板改用 DirectReplyToMessageListenerContainer,使消费者可以复用。模板仍然负责关联回复,因此不存在迟到回复被送到另一个发送者的风险。如果你想恢复以前的行为,请将 useDirectReplyToContainer(使用 XML 配置时为 direct-reply-to-container)属性设置为 false。

AsyncRabbitTemplate 没有这样的选项。当使用直接回复时,它总是使用 DirectReplyToContainer 来回复。

注意,容器在每次进行中的交换期间持有一个 channel;当连接工厂池化 channel 时这一点很重要。参见 PooledChannelConnectionFactory。

从 2.3.7 版本开始,模板有了一个新属性 useChannelForCorrelation。当它为 true 时,服务器不必把关联 ID 从请求消息头复制到回复消息。相反,用于发送请求的 channel 会被用来将回复关联到请求。

使用回复队列进行消息关联​

使用固定回复队列时(amq.rabbitmq.reply-to 除外),你必须提供关联数据,以便将回复关联到请求。参见 RabbitMQ Remote Procedure Call (RPC)。默认情况下,标准的 correlationId 属性用于保存关联数据。不过,如果你希望使用自定义属性保存关联数据,可以在 <rabbit-template/> 上设置 correlation-key 属性。显式将该属性设置为 correlationId 与省略该属性效果相同。客户端和服务器必须使用同一个头来传递关联数据。

备注

Spring AMQP 1.1 版本为此数据使用了一个名为 spring_reply_correlation 的自定义属性。如果你希望在当前版本中恢复此行为(也许是为了与另一个使用 1.1 的应用程序保持兼容),必须将该属性设置为 spring_reply_correlation。

默认情况下,模板生成自己的关联 ID(忽略任何用户提供的值)。如果你希望使用自己的关联 ID,请将 RabbitTemplate 实例的 userCorrelationId 属性设置为 true。

important

关联 ID 必须唯一,以避免某个请求收到错误回复的可能。

回复监听器容器​

当使用 3.4.0 之前版本的 RabbitMQ 时,每次回复都会使用一个新的临时队列。不过,可以在模板上配置单个回复队列,这样效率更高,还可以在该队列上设置参数。但在这种情况下,你还必须提供一个 <reply-listener/> 子元素。该元素为回复队列提供一个监听器容器,模板即为监听器。<listener-container/> 上允许的所有消息监听器容器配置属性在该元素上也都允许,除了 connection-factory 和 message-converter,它们从模板的配置继承。

important

如果你运行应用程序的多个实例,或使用多个 RabbitTemplate 实例,必须为每个实例使用唯一的回复队列。RabbitMQ 无法从队列中选择消息,因此如果它们都使用同一个队列,每个实例都会争抢回复,并且不一定能收到属于自己的回复。

下例定义了一个带有连接工厂的 rabbit 模板:

<rabbit:template id="amqpTemplate"
connection-factory="connectionFactory"
reply-queue="replies"
reply-address="replyEx/routeReply">
<rabbit:reply-listener/>
</rabbit:template>

虽然容器和模板共享同一个连接工厂,但它们不共享 channel。因此,请求和回复不在同一个事务内执行(如果是事务性的)。

备注

在 1.5.0 版本之前,没有 reply-address 属性。回复总是使用默认交换机路由,并以 reply-queue 名称作为路由键。这仍然是默认行为,但现在你可以指定新的 reply-address 属性。reply-address 可以包含 <exchange>/<routingKey> 形式的地址,回复会被路由到指定交换机,并以路由键路由到绑定的队列。reply-address 优先于 reply-queue。当只使用 reply-address 时,<reply-listener> 必须配置为单独的 <listener-container> 组件。reply-address 和 reply-queue(或 <listener-container> 上的 queues 属性)在逻辑上必须指向同一个队列。

通过此配置,使用一个 SimpleListenerContainer 接收回复,RabbitTemplate 作为 MessageListener。使用 <rabbit:template/> 命名空间元素定义模板时(如上例所示),解析器会定义容器并把模板装配为监听器。

备注

当模板不使用固定 replyQueue(或使用直接回复——参见RabbitMQ 直接回复)时,不需要监听器容器。使用 RabbitMQ 3.4.0 或更高版本时,直接 reply-to 是首选机制。

如果你把 RabbitTemplate 定义为 <bean/>,或使用 @Configuration 类将其定义为 @Bean,或者以编程方式创建模板,则需要自己定义并装配回复监听器容器。如果不这样做,模板永远收不到回复,最终会超时并在 sendAndReceive 方法调用时返回 null。

从 1.5 版本开始,RabbitTemplate 会检测自己是否已配置为接收回复的 MessageListener。如果没有,尝试带回复地址发送和接收消息会因 IllegalStateException 而失败(因为回复永远收不到)。

此外,如果使用简单的 replyAddress(队列名),回复监听器容器会验证它正在监听同名队列。如果回复地址是交换机加路由键,则无法执行此检查,并会写出一条 debug 日志消息。

important

自己装配回复监听器和模板时,务必确保模板的 replyAddress 与容器的 queues(或 queueNames)属性指向同一个队列。模板会把回复地址插入出站消息的 replyTo 属性。

下面的清单展示了手动装配 bean 的示例:

<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、并配合一个处理请求并返回回复的“远程”监听器容器的完整示例,参见此测试用例。

important

当回复超时时(replyTimeout),sendAndReceive() 方法返回 null。

在 1.3.6 版本之前,超时消息的迟到回复只会被记录日志。现在,如果收到迟到回复,它会被拒绝(模板抛出 AmqpRejectAndDontRequeueException)。如果回复队列被配置为把被拒绝的消息发送到死信交换机,该回复可以被取出供以后分析。为此,用一个路由键(等于回复队列名)将队列绑定到配置的死信交换机。

关于配置死信的更多信息,参见 RabbitMQ Dead Letter Documentation。你也可以查看 FixedReplyQueueDeadLetterTests 测试用例作为示例。

异步 Rabbit 模板​

1.6 版本引入了 AsyncRabbitTemplate。它具有与 AmqpTemplate 上类似的 sendAndReceive(和 convertSendAndReceive)方法。不过,它们不会阻塞,而是返回一个 CompletableFuture。

sendAndReceive 方法返回 RabbitMessageFuture。convertSendAndReceive 方法返回 RabbitConverterFuture。

你可以稍后通过在 future 上调用 get() 同步获取结果,也可以注册一个回调,该回调会在有结果时被异步调用。下面的清单展示了这两种方式:

@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,该异常封装了被退回的消息及退回相关信息。

如果设置了 enableConfirms,future 有一个名为 confirm 的属性,它本身是一个 CompletableFuture<Boolean>,true 表示发布成功。如果确认 future 为 false,RabbitFuture 还有一个名为 nackCause 的属性,其中包含失败原因(如果有)。

important

如果发布确认在回复之后才收到,它会被丢弃,因为回复本身意味着发布成功。

你可以在模板上设置 receiveTimeout 属性来为回复设置超时(默认为 30000,即 30 秒)。如果发生超时,future 会以 AmqpReplyTimeoutException 完成。

模板实现了 SmartLifecycle。当有待处理的回复时停止模板,会导致待处理的 Future 实例被取消。

从 2.0 版本开始,异步模板现在支持直接回复,而不是配置好的回复队列。要启用此功能,使用以下构造函数之一:

public AsyncRabbitTemplate(ConnectionFactory connectionFactory, String exchange, String routingKey)

public AsyncRabbitTemplate(RabbitTemplate template)

要在同步 RabbitTemplate 上使用直接回复,参见RabbitMQ 直接回复。

2.0 版本引入了这些方法的变体(convertSendAndReceiveAsType),它们额外接收一个 ParameterizedTypeReference 参数,用于转换复杂的返回类型。你必须为底层 RabbitTemplate 配置 SmartMessageConverter。更多信息参见用 RabbitTemplate 从消息转换。

使用 AMQP 的 Spring Remoting​

不再支持 Spring remoting,因为该功能已从 Spring Framework 中移除。

请改用 RabbitTemplate 的 sendAndReceive 操作(客户端侧)和 @RabbitListener。