spring - 如何在 Spring AMQP 中使用响应式(Reactive) RabbitListener Ack/Nack？

Question

我正在使用 Spring AMQP 2.1.6 来使用带有 RabbitListener 的消息返回 Mono<Void> .例如:

@RabbitListener
public Mono<Void> myListener(MyMessage myMessage) {
    Mono<Void> mono = myService.doSomething(myMessage);
    return mono;
}

阅读 documentation它说:

The listener container factory must be configured with AcknowledgeMode.MANUAL so that the consumer thread will not ack the message; instead, the asynchronous completion will ack or nack the message when the async operation completes.

我这样配置了容器工厂 AcknowledgeMode.MANUAL ，但我不清楚“异步完成将在异步操作完成时确认或取消消息”意味着这是由 spring-amqp 处理的。本身或者如果这是我必须做的事情？ IE。我是否必须在拨打 myService.doSomething(myMessage) 后确认/取消消息？或者 Spring AMQP 会自动确认它，因为我返回了 Mono (即使设置了 AcknowledgeMode.MANUAL)？

如果我需要手动发送 ack 或拒绝，那么在使用 RabbitListener 时以非阻塞方式执行此操作的惯用方法是什么？ ?

Answer 1

当 Mono 完成时，监听器适配器会处理 ack。

见 AbstractAdaptableMessageListener.asyncSuccess()和 asyncFailure() .

编辑

我不是 react 堆的人，但据我所知，我正在完成一个 Mono<Void>什么都不做，所以on...()方法永远不会被调用。

您可以使用 channel.basicAck 手动确认交货或 basicReject ...

@RabbitListener(queues = "foo")
public void listen(String in, Channel channel,
        @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
    ...
}