问题描述

这是我尝试在 Camel 中设置的流程:

Here is the flow I'm trying to setup in Camel:

GET/product/foo --> MULTICAST [HTTP URI 1, HTTP URI 2, HTTP URI 3] --> AGGREGATE --> 将聚合值返回到 HTTP 响应正文

GET /product/foo --> MULTICAST [HTTP URI 1, HTTP URI 2, HTTP URI 3] --> AGGREGATE --> return aggregated value to HTTP response body

我已经以这种方式设置了路由，但是在对原始 GET 的响应中没有得到任何数据.

I've setup the route this way, BUT I get no data back in the response to the original GET.

如何获取聚合器返回的值?

How can I get the value returned by the aggregator ?

 @Override
    public void configure() throws Exception {

        restConfiguration()
                .host("localhost")
                .port("8081")
                .component("jetty");

        from("rest:get:product/foo")
                .multicast()
                .parallelProcessing()
                .aggregationStrategy(new ProductPriceAggregator())
                .to("direct:prodcutService1")
                .to("direct:prodcutService2")
                .to("direct:prodcutService3");

        from("direct:prodcutService1")
                .to("http4:localhost:9090/simple/product/foo?bridgeEndpoint=true")
                .to("direct:aggregate");

        from("direct:prodcutService2")
                .to("http4:localhost:9091/simple/product/foo?bridgeEndpoint=true")
                .to("direct:aggregate");

        from("direct:prodcutService3")
                .to("http4:localhost:9092/simple/product/foo?bridgeEndpoint=true")
                .to("direct:aggregate");

        from("direct:aggregate")
                .log("${body}").;
    }
}

这是我的聚合器:

public class ProductPriceAggregator implements AggregationStrategy {

    @Override
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        System.out.println("FOO BAR");
        double oldPrice = oldExchange.getIn().getBody(Double.class);
        double newPrice = newExchange.getIn().getBody(Double.class);
        double finalPrice = oldPrice > newPrice ? newPrice : oldPrice;
        oldExchange.getIn().setBody(finalPrice);
        return oldExchange;
    }
}

推荐答案

这样就可以了.

from("direct:aggregate").transform().body(); 

但是您的聚合策略有一个小错误.这里重写了.

But there is a small mistake in your aggregation strategy. Have rewritten it here.

public class ProductPriceAggregator implements AggregationStrategy {

@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) 
{

    if (oldExchange == null) {
       // the first time we aggregate we only have the new exchange,
       // so we just return it
       return newExchange;
    }

    System.out.println("FOO BAR");
    double oldPrice = oldExchange.getIn().getBody(Double.class);
    double newPrice = newExchange.getIn().getBody(Double.class);
    double finalPrice = oldPrice > newPrice ? newPrice : oldPrice;
    oldExchange.getIn().setBody(finalPrice);
    return oldExchange;
  }
}

对于第一次迭代，oldExchange 将为 null，因此您需要检查并返回 newExchange.

For the first iteration oldExchange will be null so you need have a check and return newExchange.

出于某种奇怪的原因(或者可能是这样设计的)Camel 对待 Double 值完全不同.要使其工作，请进行以下更改.

For some strange reason(or may be it is designed like this) Camel is treating Double values completely different. To make it work, do the following changes.

 from("rest:get:product/foo")
                .setHeader("Accept", simple("application/json"))
                .multicast()
                .parallelProcessing()
                .......

这是因为，默认情况下，它采用 text/html 作为 Accept 类型，并且 double 值类似于 <Double>2.345<Double> 之类的 html 标签.因此，您需要将类型指定为 application/json 以便更好地处理.

This is because, by default it takes text/html as Accept type and double values are coming like html tags like <Double>2.345<Double>. So you need to specify type as application/json for better processing.

在聚合器代码中，您需要这样做.

And in the Aggregator code you need to do like this.

double oldPrice = Double.valueOf(oldExchange.getIn().getBody(String.class));
double newPrice = Double.valueOf(newExchange.getIn().getBody(String.class));
double finalPrice = oldPrice > newPrice ? newPrice : oldPrice;
    oldExchange.getIn().setBody(Double.toString(finalPrice));