java - 在 Camel 中设置 REST 响应主体

Question

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

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

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

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

 @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;
    }
}

Answer 1

这样就可以了。

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

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

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将为空，因此您需要检查并返回 newExchange 。

编辑:

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

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

这是因为，默认情况下它接受 text/html 作为 Accept 类型，并且 double 值就像 html 标签一样 <Double>2.345<Double> 。所以需要指定type为application/json以便更好的处理。

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

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));