使用基于计时的PollingConsumer到直接endpoint

Question

从功能上讲，我希望在从 JMS （WMQ） endpoint使用之前检查 URL 是否处于活动状态。
如果无法访问 URL 或服务器错误，则我不想从队列中选取。所以我想继续尝试（无限次重试）通过轮询消费者的 URL。因此，一旦可用，我就可以从JMS中获取。

我有一个使用直接endpoint设置的RouteBuilder，它被配置为运行将ping服务的处理器。

所以:

public class PingRoute extends RouteBuilder {
        @Override
        public void configureCamel() {
            from("direct:pingRoute").routeId(PingRoute.class.getSimpleName())
                .process(new PingProcessor(url))
                .to("log://PingRoute?showAll=true");
        }
    }

在另一条路线上，我设置了我的计时器：

    @Override
public void configureCamel() {
       from(timerEndpoint).beanRef(PollingConsumerBean.class.getSimpleName(), "checkPingRoute");
   ...
    }

使用PollingConsumerBean，我试图通过消费者接收身体：

public void checkPingRoute(){
    // loop to check the consumer.  Check we can carry on with the pick up from the JMS queue.
    while(true){
        Boolean pingAvailable = consumer.receiveBody("direct:pingRoute", Boolean.class);
   ...
    }

我将路由添加到上下文，并使用生产者发送：

context.addRoutes(new PingRoute());
context.start();
producer.sendBody(TimerPollingRoute.TIMER_POLLING_ROUTE_ENDPOINT, "a body");

我得到以下非法参数异常：

Cannot add a 2nd consumer to the same endpoint. Endpoint Endpoint[direct://pingRoute] only allows one consumer.

有没有办法将直接路由设置为轮询消费者？

Answer 1

我很难弄清楚你到底想做什么，但是在我看来，你想在一段时间内使用来自endpoint的数据。对此，最好的模式是轮询消费者：http://camel.apache.org/polling-consumer.html

您当前收到的错误是因为您有两个使用者都试图从“direct://pingRoute”中读取数据。如果这是有意的，您可以将direct更改为seda://pingRoute，这样它就是您的数据所在的内存队列。

Answer 2

根据OP对其用例的澄清，他们有几个问题需要解决：

< li >当且仅当对URL的ping是肯定的时，使用JMS队列中的消息。 < li >如果URL没有响应，JMS消息应该不会从队列中消失，必须进行重试，直到URL再次有响应，在这种情况下，消息将最终被使用。 < Li > OP没有指定重试次数是有限的还是无限的。

基于这个问题场景，我建议重新设计他们的解决方案，利用骆驼中的ActiveMQ重试、代理端重新交付和JMS事务：

1. 如果 URL ping 失败（通过事务回滚），则将消息返回到队列。
2. 确保消息不会丢失（通过使用 JMS 持久性和代理端重新传递，AMQ 将持久安排重试周期）。
3. 能够为每个消息指定复杂的重试周期，例如指数退避、最大重试次数等。
4. 如果重试周期已用尽而没有正结果，则可以选择将消息发送到死信队列，以便可以计划其他一些（可能是手动）操作。

现在，在实现方面：

from("activemq:queue:abc?transacted=true")          // (1)
    .to("http4://host.endpoint.com/foo?method=GET") // (2) (3)
    .process(new HandleSuccess());                  // (4)

评论:

1. 请注意事务标志。
2. 如果 HTTP 调用失败，HTTP4 终结点将引发异常。
3. 由于没有配置异常处理程序，Camel会将异常传播到消费方endpoint（activemq），这将回滚事务。
4. 如果调用成功，流将继续，交换正文现在将包含 HTTP 服务器返回的有效负载，您可以按照任何您希望的方式处理它。这里我使用的是处理器。

接下来，重要的是在ActiveMQ中配置重新交付策略，并启用代理端重新交付。您可以在activemq中执行此操作。xml配置文件：

<plugins>
  <redeliveryPlugin fallbackToDeadLetter="true" sendToDlqIfMaxRetriesExceeded="true">
    <redeliveryPolicyMap>
      <redeliveryPolicyMap>
        <redeliveryPolicyEntries>
          <redeliveryPolicy queue="my.queue" 
                            initialRedeliveryDelay="30000" 
                            maximumRedeliveries="17" 
                            maximumRedeliveryDelay="259200000" 
                            redeliveryDelay="30000" 
                            useExponentialBackOff="true"
                            backOffMultiplier="2" />
        </redeliveryPolicyEntries>
      </redeliveryPolicyMap>
    </redeliveryPolicyMap>
  </redeliveryPlugin>
</plugins>

并确保在顶层启用了调度程序支持

<broker xmlns="http://activemq.apache.org/schema/core" 
        brokerName="mybroker" 
        schedulerSupport="true">
    ...
</broker>

我希望那有帮助。

编辑1: OP使用IBM WebSphere MQ作为代理，我错过了这一点。您可以使用JMS QueueBrowser来查看消息，并在实际使用消息之前尝试它们对应的URL，但是不可能有选择地使用单个消息

因此，我坚持认为您应该探索JMS事务，但不是让代理重新传递消息，您可以在TX主体本身中启动对URL的ping循环。关于Camel，您可以实现如下：

from("jms:queue:myqueue?transacted=true")
    .bean(new UrlPinger());

UrlPinger.java：

public class UrlPinger {

    @EndpointInject
    private ProducerTemplate template;

    private Pattern pattern = Pattern.compile("^(http(?:s)?)\\:");

    @Handler
    public void pingUrl(@Body String url, CamelContext context) throws InterruptedException {
        // Replace http(s): with http(s)4: to use the Camel HTTP4 endpoint.
        Matcher m = pattern.matcher(url);
        if (m.matches()) {
            url = m.replaceFirst(m.group(1) + "4:");
        }

        // Try forever until the status code is 200.
        while (getStatusCode(url, context) != 200) {
            Thread.sleep(5000);
        }
    }

    private int getStatusCode(String url, CamelContext context) {
        Exchange response = template.request(url + "?method=GET&throwExceptionOnFailure=false", new Processor() {
            @Override public void process(Exchange exchange) throws Exception {
                // No body since this is a GET request.
                exchange.getIn().getBody(null);
            }
        });

        return response.getIn().getHeader(Exchange.HTTP_RESPONSE_CODE, Integer.class);
    }

}

注意事项:

< li >请注意< code > throwExceptionOnFailure = false 选项。不会引发异常，因此循环将一直执行，直到条件为真。 < li >在bean内部，我一直循环，直到HTTP状态为200。当然，你的逻辑会不一样。 < li >在尝试和尝试之间，我睡了5000毫秒。 < li >我假设ping的URL在传入JMS消息的正文中。我将前导的< code>http(s):替换为< code>http(s)4:，以便使用Camel HTTP4endpoint。 < li >在TX内部执行ping操作可以保证只有当ping条件为真时(在本例中为HTTP status == 200)才会使用该消息。 < li >您可能希望引入一个终止条件(您不想永远尝试下去)。也许引入一些后退，以不压倒对方。 < li >如果Camel或代理在重试周期内关闭，消息将自动回滚。 < li >考虑到JMS事务是与< code >会话绑定的，因此如果要启动许多并发使用者(< code > concurrent consumers JMSendpoint选项)，您需要为每个线程设置< code > CACHE level name = CACHE _ NONE 以使用不同的JMS 会话。

Answer 3

不幸的是，商业逻辑并不十分清晰。据我所知，您需要等待服务的响应。IMHO您必须使用Content Enricher EIPhttp://camel.apache.org/content-enricher.html.＜code＞pollEnrich＜/code＞是计时器路由所需的。

<代码>。pollEnrich（“direct:waitForResponce”，-1）或。pollEnrich（“seda:waitForResponce”，-1）

public class PingRoute extends RouteBuilder {
        @Override
        public void configureCamel() {
             from("direct:pingRoute").routeId(PingRoute.class.getSimpleName())
                .process(new PingProcessor(url))
             .choice().when(body())                       
                  .to("log://PingRoute?showAll=true")
                  .to("direct:waitForResponce") 
                .otherwise()
                  .to("direct:pingRoute")
                .end(); 
        }
};

定时器：

    @Override
    public void configureCamel() {                           
      from(timerEndpoint)
      .inOnly("direct:pingRoute")
      .pollEnrich("direct:waitForResponce", -1)
       ...
    }