Apache Camel：如何并行发送两个http请求并等待响应？

Question

在 Apache Camel 中，我定义了一个路由，如何并行发送两个或多个 http 请求并等待它们的“未来”以获取响应以进行进一步处理，就像在 Java 中使用 AsyncHttpClient 一样？

AsyncHttpClient asyncHttpClient = new DefaultAsyncHttpClient();
Future<Response> f = asyncHttpClient.prepareGet("http://www.example.com/").execute();
Response r = f.get();

仅作为上下文，以下路由调用 GET 联系人 http 调用并同步返回响应。

from("direct:getContact")
.to("http://host:port/contacts/1453")

Answer 1

尝试将您的路线分成许多较小的路线。然后您可以在那里执行必要的解组。

参见有关解组http响应的问题

from("direct:getContact")
    .process(new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
            CamelContext context = exchange.getContext();
            ProducerTemplate producerTemplate = context.createProducerTemplate();

            // Asynchronous call to internal route
            Future<Contact> contact = 
                producerTemplate.asyncRequestBody("direct:invokeSomeRestApi", null, Contact.class);  

            // Do rest of the work
            exchange.getOut().setBody(contact.get());
        }
    });

// Unmarshalling REST response
JacksonDataFormat jacksonDataFormat = new JacksonDataFormat();
jacksonDataFormat.setUnmarshalType(Contact.class);

// Internal route definition
from("direct:invokeSomeRestApi")
    .to("http://localhost:8080/api/contact/2345")
    .unmarshal(jacksonDataFormat);

Answer 2

您可以将骆驼多播与并行处理和聚合策略

一起使用

这个例子：

.multicast()
 .executorService(executorService)
 .parallelProcessing()
 .stopOnException()
 .aggregationStrategy(new SearchDocumentAggregator())
 .to(direct(EXEC_DOCUMENT_FULLTEXT_SEARCH))
 .to(direct(EXEC_SEARCH_DOCUMENT_METADATA))

.end()

并行执行全文搜索和元数据搜索，并且 SearchDocumentAggregator 聚合（求和）结果

公共类 SearchDocumentAggregatorimplements AggregationStrategy { 公共交换聚合（交换原始、交换资源）{

    if(newExchange.getIn().getHeader(ServiceUtils.RESPONSE_FROM, String.class).equals(EnumDppResponseFrom.SEARCH_FULLTEXT.name())) {
       //here get the fulltext response to SearchAggregator
    }
    if(newExchange.getIn().getHeader(ServiceUtils.RESPONSE_FROM, String.class).equals(EnumDppResponseFrom.SEARCH_METADATA.name())) {
         //here get the metadata response and add to SearchAggregator
    }

    newExchange.getIn().setBody(fullTextSearchAggregator, SearchAggregator.class);


     return newExchange;
}

}