如何窃听rabbitmq rpc响应消息

问题描述

MessageListenerContainer 和 MessageListenerAdapter 实现为

void handleMessage(MyRpcRequest request,MessageProperties messageProperties) {
        var correlationData = new CorrelationData(messageProperties.getCorrelationId());
        MessagePostProcessor messagePostProcessor = (m) -> {
            m.getMessageProperties().setCorrelationId(correlationData.getId());
            if (messagePostProcessorFactory != null) {
                messagePostProcessorFactory.create(exchange,routingKey).postProcessMessage(m);
            }
            return m;
        };
        MyRpcResponse response = computeResponse(request);
    rabbitTemplate.convertAndSend("",messageProperties.getReplyTo(),response,messagePostProcessor,correlationData);
}

我直接发送回默认交换 - 它工作正常 - 接收到 rpc 响应（更简单的是返回响应，没有用兔子模板发送回，但我故意手动发送）。但是这样我就无法窃听响应消息。我想做的是像下面这样 - 发送到 RESPONSE_EXCHANGE_WHICH_IS_BOUND_TO_DEFAULT_EXCHANGE 这是扇出交换 - 我可以从它窃听消息，然后我希望它转发消息到默认交换 - 我尝试将我的响应交换绑定到默认交换，但是它不起作用

void handleMessage(MyRpcRequest request,MessageProperties messageProperties) {
    ....
    rabbitTemplate.convertAndSend("RESPONSE_EXCHANGE_WHICH_IS_BOUND_TO_DEFAULT_EXCHANGE",correlationData); 
    ....
}

    @Bean
    FanoutExchange defaultExchange() {
        return new FanoutExchange("");
    }

    @Bean
    FanoutExchange myRpcResponseExchange() {
        return new FanoutExchange("RESPONSE_EXCHANGE_WHICH_IS_BOUND_TO_DEFAULT_EXCHANGE");
    }

    @Bean
    Binding bindMyRpcResponseExchangetoDefaultExchange(FanoutExchange myRpcResponseExchange,FanoutExchange defaultExchange) {
        return BindingBuilder.bind(myRpcResponseExchange).to(defaultExchange);
    }

如何修复我的代码以便我能够窃听响应消息？

我如何窃听响应消息发送的请求：

rabbitTemplate.convertSendAndReceiveAsType(exchange,routingKey,message,correlationData,responseType); ?

解决方法

您不得手动将任何内容绑定到默认交换。

查看管理 UI - 没有“绑定”选项。

你也不能将默认交换绑定到另一个。

您需要发布到扇出并将回复队列绑定到它。