apache骆驼轮询休息端点

Posted 2023-02-23

【中文标题】apache骆驼轮询休息端点

【英文标题】：apache camel polling a rest endpoint

【发布时间】：2016-11-23 09:50:47

【问题描述】：

我有一个休息端点 sample.org，它返回格式为 json 的响应

  "response" : "pending" 

我的路线是这样的

from("http://sample.org")
.marshal(xmlFormatConverterUtil.getxmlJsonDataFormat())  //To convert into json as I receive data in xml format which needs to be converted to json

我阅读了有关 polling consumer 的信息，但找不到有关如何继续轮询端点的示例，直到它返回响应为“成功”。

应该使用轮询消费者吗？如果是这样，可以说明与我的案例相关的示例。任何其他轮询休息端点的资源都将非常有用。

【问题讨论】：

轮询 http 端点是什么意思？该路由正在侦听暴露的 http url。每当有人向该网址发送请求时，您都会收到该消息。这里没有投票。

通过轮询我的意思是，我将继续向端点发出 GET 请求，并且每次我都会检查我的响应是什么。当响应成功时，我将停止轮询端点

【参考方案1】：

您需要从计时器开始，然后调用其余端点。然后您可以检查结果，如果结果，则使用 controlbus 停止路线。过滤器可用于检查其是否挂起，然后停止继续路由，然后下一个计时器将重试。

沿着这条伪路线前进

from timer
  to http
  marshal
  filter (if pending)
     stop 
  end
  to something with positive response
  to controlbus stop route

您可以在

找到更多详细信息 http://camel.apache.org/timer http://camel.apache.org/controlbus http://camel.apache.org/how-can-i-stop-a-route-from-a-route.html http://camel.apache.org/message-filter.html

【讨论】：

新书怎么样了？我的第 1 版副本全是狗耳朵的。

我们有 2.5 章要写，Henryk 会写 IoC 章。但是从明年开始，我们已经完成了所有这些，然后是它的证明，排版。在 2017 年夏天之前完成。

【参考方案2】：

我遇到了类似的问题，最后通过writing a custom endpoint 进行投票。

它作为生产者工作并轮询指定的 uri，直到满足指定的谓词或轮询达到最大尝试次数。

from("direct:start")
  .to("poll:http://example.com/status?maxRetries=3&successPredicate=#statusSuccess")

轮询端点使用一个简单的处理器，该处理器使用轮询消费者进行轮询。

public class PollProcessor implements Processor 

    private final String uri;
    private final long requestTimeoutMs;
    private final long period;
    private final int maxTries;
    private final Predicate<Exchange> successPredicate;

    public PollProcessor(String uri, long requestTimeoutMs, long period, int maxTries, Predicate<Exchange> successPredicate) 
        Preconditions.checkArgument(maxTries > 0);
        Preconditions.checkArgument(period >= 0);
        Preconditions.checkNotNull(successPredicate);

        this.uri = uri;
        this.requestTimeoutMs = requestTimeoutMs;
        this.period = period;
        this.maxTries = maxTries;
        this.successPredicate = successPredicate;
    

    @Override
    public void process(Exchange exchange) throws Exception 
        PollingConsumer consumer = exchange.getContext().getEndpoint(uri).createPollingConsumer();

        for (int tryNumber = 1; tryNumber <= maxTries; ++tryNumber) 
            Exchange pollExchange = consumer.receive(requestTimeoutMs);
            if (successPredicate.test(pollExchange)) 
                exchange.setOut(pollExchange.getOut());
                exchange.setException(pollExchange.getException());
                return;
            

            log.warn("Polling  failed try number , waiting  ms for next try...", uri, tryNumber);
            Thread.sleep(period);
        

        throw new RuntimeException("Polling failed maximum allowed number of tries [" + maxTries + "], see log for details.");
    

【讨论】：

很好：D 我通过在我的路线中循环使用计时器和重试计数来解决它，类似于你所做的。