回复管理

MessageListenerAdapter 中现有的支持已允许您的方法具有非 void 返回类型。在这种情况下，调用的结果将封装在发送到原始消息的 ReplyToAddress 标头中指定的地址或监听器上配置的默认地址的消息中。您可以使用消息传递抽象的 @SendTo 注解来设置该默认地址。

假设我们的 processOrder 方法现在应该返回 OrderStatus，我们可以按如下方式编写它以自动发送回复

@RabbitListener(destination = "myQueue")
@SendTo("status")
public OrderStatus processOrder(Order order) {
    // order processing
    return status;
}

如果您需要以独立于传输的方式设置其他标头，您可以返回 Message，如下所示

@RabbitListener(destination = "myQueue")
@SendTo("status")
public Message<OrderStatus> processOrder(Order order) {
    // order processing
    return MessageBuilder
        .withPayload(status)
        .setHeader("code", 1234)
        .build();
}

或者，您可以在 beforeSendReplyMessagePostProcessors 容器工厂属性中使用 MessagePostProcessor 添加更多标头。从 2.2.3 版本开始，调用的 bean/方法在回复消息中可用，可用于消息后处理器将信息传回给调用方

factory.setBeforeSendReplyPostProcessors(msg -> {
    msg.getMessageProperties().setHeader("calledBean",
            msg.getMessageProperties().getTargetBean().getClass().getSimpleName());
    msg.getMessageProperties().setHeader("calledMethod",
            msg.getMessageProperties().getTargetMethod().getName());
    return m;
});

从 2.2.5 版本开始，您可以配置 ReplyPostProcessor 以在发送回复消息之前对其进行修改；它在设置 correlationId 标头以匹配请求后被调用。

@RabbitListener(queues = "test.header", group = "testGroup", replyPostProcessor = "echoCustomHeader")
public String capitalizeWithHeader(String in) {
    return in.toUpperCase();
}

@Bean
public ReplyPostProcessor echoCustomHeader() {
    return (req, resp) -> {
        resp.getMessageProperties().setHeader("myHeader", req.getMessageProperties().getHeader("myHeader"));
        return resp;
    };
}

从 3.0 版本开始，您可以在容器工厂上配置后处理器，而不是在注解上配置。

factory.setReplyPostProcessorProvider(id -> (req, resp) -> {
    resp.getMessageProperties().setHeader("myHeader", req.getMessageProperties().getHeader("myHeader"));
    return resp;
});

id 参数是监听器 ID。

注解上的设置将覆盖工厂设置。

@SendTo 值被假定为遵循 exchange/routingKey 模式的回复 exchange 和 routingKey 对，其中可以省略其中一部分。有效值为：

thing1/thing2：replyTo 交换机和 routingKey。thing1/：replyTo 交换机和默认（空）routingKey。thing2 或 /thing2：replyTo routingKey 和默认（空）交换机。/ 或空：replyTo 默认交换机和默认 routingKey。

此外，您可以在没有 value 属性的情况下使用 @SendTo。此情况等同于空 sendTo 模式。仅当传入消息没有 replyToAddress 属性时才使用 @SendTo。

从 1.5 版本开始，@SendTo 值可以是 bean 初始化 SpEL 表达式，如下例所示

@RabbitListener(queues = "test.sendTo.spel")
@SendTo("#{spelReplyTo}")
public String capitalizeWithSendToSpel(String foo) {
    return foo.toUpperCase();
}
...
@Bean
public String spelReplyTo() {
    return "test.sendTo.reply.spel";
}

表达式必须计算为 String，可以是简单的队列名称（发送到默认交换机）或采用 exchange/routingKey 格式，如前面的示例中所述。

#{…} 表达式在初始化期间评估一次。

对于动态回复路由，消息发送方应包含 reply_to 消息属性或使用备用运行时 SpEL 表达式（在下一个示例之后描述）。

从 1.6 版本开始，@SendTo 可以是根据请求和回复在运行时计算的 SpEL 表达式，如下例所示

@RabbitListener(queues = "test.sendTo.spel")
@SendTo("!{'some.reply.queue.with.' + result.queueName}")
public Bar capitalizeWithSendToSpel(Foo foo) {
    return processTheFooAndReturnABar(foo);
}

运行时 SpEL 表达式的性质由 !{…} 定界符指示。表达式的评估上下文 #root 对象具有三个属性

request：o.s.amqp.core.Message 请求对象。
source：转换后的 o.s.messaging.Message<?>。
result：方法结果。

上下文具有映射属性访问器、标准类型转换器和 bean 解析器，允许引用上下文中的其他 bean（例如，@someBeanName.determineReplyQ(request, result)）。

总之，#{…} 在初始化期间评估一次，#root 对象是应用程序上下文。通过其名称引用 bean。!{…} 针对每条消息在运行时评估，根对象具有前面列出的属性。通过其名称引用 bean，并在其前面加上 @。

从 2.1 版本开始，还支持简单的属性占位符（例如，${some.reply.to}）。在早期版本中，可以使用以下内容作为变通方法，如下例所示

@RabbitListener(queues = "foo")
@SendTo("#{environment['my.send.to']}")
public String listen(Message in) {
    ...
    return ...
}