Apache camel 聚合多个 REST 服务响应

Posted 2023-03-10

【中文标题】Apache camel 聚合多个 REST 服务响应

【英文标题】：Apache camel to aggregate multiple REST service responses

【发布时间】：2012-05-23 13:26:30

【问题描述】：

我是 Camel 的新手，想知道如何使用 Camel 实现下面提到的用例，

我们有一个 REST Web 服务，假设它有两个服务操作 callA 和 callB。 现在我们在前面有 ESB 层，它在访问这个实际的 Web 服务 URL 之前拦截客户端请求。

现在我正在尝试做这样的事情 - 在 ESB 中公开客户端将实际调用的 URL。在 ESB 中，我们使用 Camel 的 Jetty 组件，它只是代理此服务调用。所以假设这个 URL 是 /my-service/scan/

现在在收到@ESB 请求时，我想调用这两个 REST 端点（callA 和 callB）-> 获取它们的响应 - resA 和 resB -> 将其聚合到单个响应对象 resScan -> 返回到客户端。

我现在只有-

<route id="MyServiceScanRoute">
<from uri="jetty:http://host.port./my-service/scan/?matchOnUriPrefix=true&amp;bridgeEndpoint=true"/>
<!-- Set service specific headers, monitoring etc. -->  
<!-- Call performScan -->
<to uri="direct:performScan"/>
</route>

<route id="SubRoute_performScan">
<from uri="direct:performScan"/>
<!--  HOW DO I??
Make callA, callB service calls. 
Get their responses resA, resB.
Aggregate these responses to resScan
 -->
</route>

【参考方案1】：

我认为您不必要地使解决方案复杂化了一点。 :) 在我看来，调用两个独立远程 Web 服务并连接结果的最佳方法是：

使用multicast

并行

调用服务 使用GroupedExchangeAggregationStrategy 聚合结果

上述解决方案的路由可能如下所示：

from("direct:serviceFacade")
  .multicast(new GroupedExchangeAggregationStrategy()).parallelProcessing()
    .enrich("http://google.com?q=Foo").enrich("http://google.com?q=Bar")
  .end();

传递给direct:serviceFacadeResponse 的Exchange 将包含属性Exchange.GROUPED_EXCHANGE 设置为调用您的服务的结果列表（在我的示例中为Google 搜索）。

这就是你如何将direct:serviceFacade 连接到 Jetty 端点：

from("jetty:http://0.0.0.0:8080/myapp/myComplexService").enrich("direct:serviceFacade").setBody(property(Exchange.GROUPED_EXCHANGE));

现在，您使用 Jetty 组件在 ESB 上公开的对服务 URL 的所有 HTTP 请求都将生成来自对子服务的两次调用的串联响应。

关于消息和端点的动态部分的进一步考虑

在许多情况下，在端点中使用静态 URL 不足以满足您的需求。在将负载传递给每个 Web 服务之前，您可能还需要准备负载。

一般来说 - 用于实现动态端点或有效负载参数的路由类型高度依赖于您用于使用 Web 服务的组件（HTTP、CXFRS、Restlet、RSS 等）。每个组件的程度和动态配置方式各不相同。

如果您的端点/有效负载应该受到动态影响，您还可以考虑以下选项：

使用多播端点的onPrepareRef 选项预处理传递到每个端点的交换副本。您可以使用它来引用自定义处理器，该处理器将在将有效负载传递到多播端点之前对其进行修改。这可能是使用 HTTP 组件的 Exchange.HTTP_URI 标头组合 onPrepareRef 的好方法。

使用Recipient List（它也像多播一样提供parallelProcessing）来动态创建 REST 端点 URL。

使用Splitter 模式（启用parallelProcessing）将请求拆分为专用于每个服务的较小消息。再次，此选项可以很好地与 HTTP 组件的 Exchange.HTTP_URI 标头一起使用。这只有在两个子服务都可以使用相同的端点类型定义时才有效。

如您所见，Camel 非常灵活，可让您通过多种方式实现目标。考虑您的问题的背景，然后选择最适合您的解决方案。

如果您向我展示您希望针对聚合服务的每个请求调用的 REST URL 的更具体示例，我可以建议您选择哪种解决方案以及如何实现它。特别重要的是要知道请求的哪一部分是动态的。我还需要知道您要使用哪个服务使用者（这取决于您将从服务接收到的数据类型）。

【讨论】：

感谢您接受我的回答。也欢迎投票：P。

多播指令的精彩演示，Henryk。但是，似乎需要准备有效负载，而不仅仅是对每个服务的请求的 URL。你也可以用多播来做到这一点吗？不是将同一消息多播到多个端点吗？

@Petter 当我到达 REST 端点时，它几乎只是 URL（GET 请求）。但是我在这里有一个问题，我需要将参数传递给各个服务调用。例如：0.0.0.0:8080/myapp/myComplexService?keyword=abc 所以这个“关键字”参数必须在单独的服务调用中传递 -> myserver.com/serviceA?keyword=abc & myserver.com/serviceB?keyword=abc

@Rishi，Henryk 在上面证明了这一点。查看 URI，q=Foo 和 q=Bar。这将代表您的关键字=abc。

@Petter 其实想说的是——当我调用服务时，客户端传递的所有 url 参数都应该传递（结转）。我在示例中看到的是一个硬编码参数，仅用于简化演示（两个服务调用将始终使用 q=foo 和 q=bar）。阅读 Camel 文档后，我发现 http 组件在这方面很有用。这样我会将客户端发送的 URL 参数转发到不同的端点。这听起来正确吗？

【参考方案2】：

这看起来是一个很好的例子，应该使用 Content Enricher 模式。 Described here

<from uri="direct:performScan"/>
   <enrich uri="ServiceA_Uri_Here" strategyRef="aggregateRequestAndA"/>
   <enrich uri="ServiceA_Uri_Here" strategyRef="aggregateAandB"/>
</route>

聚合策略必须用 Java 编写（或者可能是某种脚本语言，Scala/groovy？ - 但我没有尝试过）。

聚合策略只需要一个实现 org.apache.camel.processor.aggregate.AggregationStrategy 的 bean，这反过来又需要你实现一个方法：

Exchange aggregate(Exchange oldExchange, Exchange newExchange);

所以，现在由您将请求与来自丰富服务调用的响应合并。你必须做两次，因为你有 callA 和 callB。有两种预定义的聚合策略可能有用也可能没用，UseLatestAggregationStrategy 和 UseOriginalAggregationStrategy。这些名称很容易解释。

祝你好运

【讨论】：

感谢指点！真的很感激。让我开始阅读内容丰富模式等。

在与团队进行一轮讨论后，我们发现 Enricher 选项的成本很高，因为它是连续的（我们可能有超过 2 个服务调用）。现在我正在评估使用的选项 - 1. 组合消息处理器 EIP/拆分器-聚合器 2. 或分散-收集 EIP 3. 或具有聚合器策略的收件人列表 任何输入？哪一个最适合用例？

问题是将请求和聚合响应保持在同一路由中，因为您要回复 HTTP 调用。最吸引人的选择似乎是指定了聚合策略且“parallellProcessing”选项设置为 true 的接收者列表。棘手的部分是编写聚合策略，因为它可能以任何顺序接收答案。我对有效载荷一无所知，但它应该相当直截了当。 camel.apache.org/aggregator.html 有一些示例如何聚合多个响应，例如响应序列中的最高数字。 gl

看起来在多播和收件人列表模式中，“相同”消息被路由到多个消费者。但在这种情况下，两个不同的 Web 服务调用（ callA 和 callB ）是完全不同的。所以我要尝试这样的事情 -