Apache Camel 超时同步路由答案

【问题标题】：Apache Camel timeout synchronous routeApache Camel 超时同步路由
【发布时间】：2018-03-15 19:50:54
【问题描述】：

我想用 Apache Camel 构建一个带超时的同步路由，但我在框架中找不到任何可以解决它的东西。所以我决定为我构建一个流程。

public class TimeOutProcessor implements Processor {

private String route;
private Integer timeout;

public TimeOutProcessor(String route, Integer timeout) {
    this.route = route;
    this.timeout = timeout;
}


@Override
public void process(Exchange exchange) throws Exception {
    ExecutorService executor = Executors.newSingleThreadExecutor();

    Future<Exchange> future = executor.submit(new Callable<Exchange>() {

        public Exchange call() {
            // Check for field rating

            ProducerTemplate producerTemplate = exchange.getFromEndpoint().getCamelContext().createProducerTemplate();
            return producerTemplate.send(route, exchange);
        }
    });
    try {
        exchange.getIn().setBody(future.get(
                timeout,
                TimeUnit.SECONDS));
    } catch (TimeoutException e) {
        throw new TimeoutException("a timeout problem occurred");
    }
    executor.shutdownNow();
}

我这样称呼这个过程：

.process(new TimeOutProcessor("direct:myRoute",
                Integer.valueOf(this.getContext().resolvePropertyPlaceholders("{{timeout}}")))

我想知道我的方式是否是推荐的方式，如果不是，构建具有超时的同步路由的最佳方式是什么？

【问题讨论】：

您的具体用例是什么？我认为没有理由对 Camel 中的直接路由应用超时。为什么需要它？
因为我的路由需要和数据库通信，而且这个路由必须是同步的，因为有必要这个路由需要在特定的时间执行。
ProducerTemplate 上有一些异步 API，您可以使用它来发送和取回 Future，然后您可以通过超时获取。
但是，我需要一个同步路由。有什么方法可以在同步中转换异步路由以使用超时？
stackoverflow.com/users/406429/claus-ibsen 的意思是您可以使用带有 ProducerTemplate 的 bean，它通过像 asyncSend 这样的异步方法调用直接路由（有关更多信息，请参阅 here）。这会立即为您提供 Java Future。在此 Future 上，您可以请求超时的结果（请参阅 get 方法，在 javadocs 中超时。您有一个同步路由，并在超时时调用它。

标签： apache-camel spring-camel

【解决方案1】：

我要感谢回答我的人。

这是我的最终代码：

public class TimeOutProcessor implements Processor {

private String route;
private Integer timeout;

public TimeOutProcessor(String route, Integer timeout) {
    this.route = route;
    this.timeout = timeout;
}


@Override
public void process(Exchange exchange) throws Exception {
    Future<Exchange> future = null;
    ProducerTemplate producerTemplate = exchange.getFromEndpoint().getCamelContext().createProducerTemplate();
    try {

        future = producerTemplate.asyncSend(route, exchange);
        exchange.getIn().setBody(future.get(
                timeout,
                TimeUnit.SECONDS));
        producerTemplate.stop();
        future.cancel(true);
    } catch (TimeoutException e) {
        producerTemplate.stop();
        future.cancel(true);
        throw new TimeoutException("a timeout problem occurred");
    }

}
}

【讨论】：