Spring集成确认消息处理回之前的服务答案

【问题标题】：Spring integration acknowledge message processing back to previous serviceSpring集成确认消息处理回之前的服务
【发布时间】：2015-01-07 19:31:21
【问题描述】：

在 Spring Integration 中，我有一个服务链，如下所示：

message -> A -> B -> C -> D -> ... -> output

这很好用。我想让每个服务都异步并使它们变得悲观。他们每个人都会收到一条消息，对其进行处理并将其发送到链中的下一个服务。但是，它不会等到整个链条完成。它将继续处理下一条消息，依此类推。此处为标准异步。

但是，假设服务 B 比 A 慢，并且它在入站通道队列中累积了 10k 条消息，此时系统崩溃。我希望能够通过找出我离开的位置并重新处理消息来恢复系统。出于这个原因，我希望每个服务都知道它处理的哪些消息被以下服务成功使用。已发送与已处理之间的区别。

我的想法是做类似这样的事情（花哨的 ascii）：

-> A --> B -> C -> ...
   ^     |
   | ack |
   \-----/

也就是说，A 将发送给 B，B 将进行处理，当它成功完成后，它会向 A 发送一个 ack。然后 A 将从存储中删除该特定消息，以便下次运行时，它会不重新处理它。我想我会在 B 之后放置一个拆分器，它会在服务 A 上调用不同的方法（即ackProcessed）。

这是应该在 SI 中完成的方式，还是我缺少另一种方式？我主要是要求确认我没有遗漏开箱即用支持的内容或不会强迫我在每个服务之后创建拆分器的内容。

【问题讨论】：

【解决方案1】：

它不会是一个分离器；更有可能是一个发布-订阅通道，而 ack 可能希望转到 A 中的不同方法（即引用相同 bean 的不同服务激活器，不同方法；并且这些方法共享某些状态）。

更简单的解决方案是使用持久消息通道（例如 JMS、RabbitMQ 或消息存储支持的 QueueChannel）。这样框架就会为你处理好一切。

【讨论】：

谢谢加里。请注意，我不在乎在关机时丢失确认。你能澄清一下使用持久消息通道会如何改变事情吗？另外，为什么分离器不起作用？我可以将消息拆分到两个目的地（在本例中为 A 和 C），可能在它和 A 之间添加一个转换器（如果拆分器也不能进行转换）。这不会按预期工作吗？
这意味着一旦A发送了它的消息，它就不再关心了，因为如果系统崩溃，当系统下次运行时，任何排队的消息仍然存在并且将由 B 处理。拆分器是将一个（例如集合）拆分为单独的消息，或者说，将订单拆分为 OrderLines。拆分器不用于向 2 个消费者发送相同的消息。
谢谢加里，现在明白你的意思了。我假设使用 pub-sub 通道之后我会添加一个桥以使其成为可轮询的，因为我希望服务被异步调用，对吗？
是的；或者您可以将 task-executor 添加到 pub-sub 频道，然后“队列”位于任务执行器中，而不是频道中。
谢谢加里 - 我会试试的！