为什么Channel.waitForConfirmsOrDie阻塞？

Question

我有一个发布 - 订阅用例,我想在发布端阻止,直到每个订阅者确认他们已经完成处理发布者发送的消息.

我(错误地？)假设我可以使用RabbitMQ及其Java amqp-client的Channel.waitForConfirmsOrDie方法作为我的解决方案的一部分.问题是我没有找到waitForConfirmsOrDie实际阻止的情况.

根据javadocs,waitForConfirmsOrDie应该:

等到自上次呼叫以来发布的所有消息都被代理确认或者没有.如果任何消息都没有,那么waitForConfirmsOrDie将抛出IOException.在非确认频道上呼叫时,它将立即返回.

为了测试这个方法是否真的有效,我从RabbitMQ网站开始使用这个示例代码.

示例代码创建了一个发布者和一个使用者,每个都在自己独立的线程上.然后,当消费者使用消息时,发布者将消息发送到交换机.似乎发布者应该阻止,直到通过调用waitForConfirmsOrDie()来获取所有消息.

这个示例代码似乎与我尝试做的完全匹配.但是,它似乎没有像我想象的那样工作.实际上,如果在消费者线程中,我关闭了自动执行消息,那么waitForConfirmsOrDie()仍会立即返回.

我通过将一个false更改为true来关闭auto ack: ch.queueDeclare(QUEUE_NAME, false, false, false, null); 变为 ch.queueDeclare(QUEUE_NAME, true, false, false, null);(2nd arg false而不是true).我相信这意味着消费者不应再发送消息.

那么waitForConfirmsOrDie()实际上做了什么？什么时候会阻止？

如果waitForConfirmsOrDie没有做我想要的,有没有办法让发布者等到所有订阅者在继续之前收到消息？

Answer 1

据我所知,这些电话不应等待消费者的确认.waitForConfirms*方法的目的是确保将消息传递给代理并提供基本传递/失败类型的通知.换句话说,如果rmq节点之一(或甚至所有节点)失败/不可用,则消息不会在没有通知产生的情况下消失.

如果在basicPublish呼叫前断开连接或关闭rmq,则可以看到此异常.