将消息从“回复”队列复制到另一个队列

Question

我主要在RPC模式下使用rabbitMq，但我还想将请求和响应消息复制到另一个队列。

最后，我想实现的是，外部消费者可以通过听一个队列来查看所有流量，我们称之为“日志队列”。

复制传入消息是可以的，我只需要使用扇出交换，或者使用与RPC调用使用的路由密钥相同的路由密钥将日志队列绑定到使用过的交换。

但我无法找到通过直接回复功能“扇出”发送的消息的方法。

到目前为止，我了解到响应消息以amqp的形式通过生成的routing_密钥发送到默认的direct exchange。rabbitmq。回复。generatedName，由于默认的交换是不可接触的，所以我不能复制这些消息。

你知道怎么做吗？

我有一个我宁愿避免的解决方案：让客户端将从回复接收到的响应重新发送到“日志队列”。

但这意味着我的客户要对这个“日志”功能负责，我宁愿不负责。

顺便说一句，即使我认为这不相关，因为这可能是rabbitMq服务器配置问题，我也使用Spring AMQP客户端

Answer 1

我已经找到了实现我想要的东西的方法。

这并不像Gary Russell提议的那样灵活：https://stackoverflow.com/a/59976806/2546702

但是，我可以利用fire hose功能并将amq.rabbitmq.trace上的队列（或交换以获得更多控制）与设置为“发布”的例程键绑定（末尾的点很重要）

这允许记录发布到默认交换的消息，包括回复消息。

当然，使用firehose会对性能产生影响，但就我而言，这并不是什么大问题，因为rabbitmq没有得到充分利用。

因为我有16个队列要听，所以我不想对每个队列使用不同的模板。我可以对所有RPC队列使用单个回复队列，但这将是一个瓶颈。

所以，如果没有硬性酒，消防水龙带似乎是一个不错的选择。

Answer 2

你不能用固定的回复来做这件事，因为没有真正的队列/交换。

但是，您可以将每个RabbitTemboard配置为使用固定的回复队列和回复容器将回复从该队列定向到模板。

文件在这里。

此外，使用此机制时，可以将模板的replyAddress配置为交换和路由密钥的形式。

/**
 * An address for replies; if not provided, a temporary exclusive, auto-delete queue will
 * be used for each reply, unless RabbitMQ supports 'amq.rabbitmq.reply-to' - see
 * https://www.rabbitmq.com/direct-reply-to.html
 * <p>The address can be a simple queue name (in which case the reply will be routed via the default
 * exchange), or with the form {@code exchange/routingKey} to route the reply using an explicit
 * exchange and routing key.
 * @param replyAddress the replyAddress to set
 */
public synchronized void setReplyAddress(String replyAddress) {...}

您只需按正常方式设置模板和容器，使模板成为容器的侦听器。。。

@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;
}

同样，这是从留档。