使用Spring Integration聚合异步结果

嗨，我遇到了一个问题，该问题使用Spring Integration具有非常好的解决方案。很多时候，我们需要一种将消息分发到未知数量的目的地的方案。为此，我们使用主题方法。但是有时，我们还希望从收到消息的所有目的地接收答案，并将其汇总为单个结果答案。为此，我们可以将通道与Aggregator和ReleaseStrategy接口结合使用。在这篇文章中，我不会专注于“渠道实施”。

假设我们有一个生产者将其消息发送给主题。现在我们有一个接收该消息的消费者。使用网关和处理器接口，我们可以以任何所需的类型发送该消息：

public interface Processor
{public void sendResponse(String response);
}

消费者代码：

@Override public void onMessage(Message message)
{String resultMessage = "";try{processor.sendResponse(resultMessage);}catch (Exception e){log.error("Error while processing message in channel consumer. errorMsg=" + e.getMessage(), e);}
}

现在，该消息将被传递到一个通道（“在通道中”）。我们可以向该消息添加额外的信息（如果我们有不同的消息组）。添加了额外的信息后，我们将该消息发送到另一个通道（“ Out通道”），这就是魔术：我们创建了两个pojo，稍后将使用XML配置将其绑定到接口。发行策略：

public class ReleaseStrategy
{public boolean canRelease(List results){// check if all 5  subscribers sent responsesif (results.size() == 5){return true;}return false;}
}

聚合器：

public class Aggregator
{public String aggregate(List results){String finalResult= "SUCCESS_RESULT";for (String result: results) {if (result.equals("ERROR_RESULT")) {finalResult= "ERROR_RESULT";break;}}return finalResult;}
}

基本上，这里发生的事情是，在我们通过ReleaseStrategy接口的canRelease方法返回“ true”值之后，聚合器将能够接收聚合的消息并将单个结果分派到最终目的地（可能是另一个消息使用者将获得的队列并处理结果）Xml配置：