在我们的应用程序中，发布者创建一条消息并将其发送到一个主题。
然后它需要等待，当所有主题的订阅者都确认消息时。
它不会出现，消息总线实现可以自动执行此操作。因此，我们倾向于让每个订阅者在完成后向客户发送他们自己的新消息。
现在，客户端可以接收所有此类消息，并且当它从每个目的地收到一条消息时，就可以进行任何它必须做的清理工作。但是，如果客户端（发送方）在确认流中途崩溃了怎么办？为了处理这样的不幸，我需要在客户端（重新）实现总线已经实现的东西——保存传入的确认，直到我得到足够的确认。
我不相信，我们的需求是如此深奥——您将如何处理这种情况，发件人（发布者）必须等待来自多个收件人（订阅者）的确认？有点像从每个订阅者向邮件列表请求（和等待）回执......
如果重要的话，我们正在使用 RabbitMQ。谢谢！
您正在寻找的功能听起来像是一种消息传递解决方案，可以跨消息的发布者和订阅者执行事务。在 Java 世界中，JMS 指定了此类事务。JMS 实现的一个示例是HornetQ。
RabbitMQ 不提供此类功能，但这样做是有充分理由的。RabbitMQ 的构建是为了非常健壮并同时表现得像地狱一样。您描述的事务行为只有以合理的性能损失为代价才能实现（特别是如果您想保持出色的稳健性）。
使用 RabbitMQ，确保消息被成功消费的一种方法确实是在消费者端发布一个应答消息，然后由原始发布者消费。这可以通过RabbitMQ 的 RPC 过程调用来实现，这可能会帮助您为您的问题设置找到一个干净的解决方案。
如果（原始）发布者在收到所有答案之前崩溃，您可以假设所有未完成的答案仍在代理中排队。因此，您必须以能够恢复处理那些留下的消息的方式构建您的发布者。这可能不是微不足道的。
最后，我推荐以下解决方案：设计您的生产组件，使您可以使用与原始发布者分离的一个或多个专用答案消费者来使用答案。
该解决方案的好处是：
现在更笼统地说：消息传递的主要好处之一是代理将应用程序组件解耦。在 AMQP 中，这是通过交换和绑定来实现的，这些交换和绑定允许您将消息分发逻辑从应用程序移动到配置的中心点。
如果您向客户端添加 RPC 样式的调用，那么您的组件很可能再次紧密耦合，这意味着如果使用组件之一失败/不可用/太慢，则发布组件将失败。这正是您想要避免的。否则，你为什么要拆分组件呢？
我的建议是，您的应用程序设计方式应使发布者可以尽可能独立于消费者的成功来完成他们的任务。反向通道应该是一个例外情况，并以所描述的非耦合方式实现。