回复管理

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 模式的回复 exchangeroutingKey 对, 其中一个部分可以省略。 有效值如下:

  • thing1/thing2: replyTo 交换和 routingKeything1/: replyTo 交换和默认(空)routingKeything2/thing2: replyTo routingKey 和默认(空)交换。 / 或空: replyTo 默认交换和默认 routingKey

此外,你可以使用不带 value 属性的 @SendTo。 这种情况等同于空的 sendTo 模式。 @SendTo 仅在入站消息没有 replyToAddress 属性时使用。

从 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 ...
}