取消 Camel 消息

Question

我支持一个应用程序，该应用程序使用 Apache Camel 作为 SOAP 和 RESTful 消息的路由中心。我们实施了一些相当复杂的路由规则。有一个用户界面，管理员可以在其中监视通过集线器的消息及其状态。我有一个增强请求，允许用户请求删除任何正在进行的请求。这通常是由于某种原因发送失败的请求，Camel 已将其排队等待重新发送。

其中一位开发人员添加了一些代码，这些代码将从飞行存储库中删除特定消息，并将我们存储库的状态更改为已取消。我发现 Camel 无论如何都会尝试重新传递该消息，直到达到最大值重试计数，消息以失败状态结束，而不是取消。

到位的代码是：

@Override
@Asynchronous
public void cancelExchange(String exchangeId) {
    synchronized (this.camelContext) {
        // Get a list of all in flight exchanges with the passed in exchangeId
        final List<Exchange> exchanges = this.camelContext.getInflightRepository().browse().stream().filter(inflightExchange -> inflightExchange.getExchange().getExchangeId().equals(exchangeId))
                .map(InflightRepository.InflightExchange::getExchange).collect(Collectors.toList());

        // Remove any active exchanges from Camel
        for (final Exchange exchange : exchanges) {
            this.log.debug("Removing exchange {}", exchange.getExchangeId());

            this.camelContext.getInflightRepository().remove(exchange);

            // Set the ROUTE_STOP property so Camel will try to stop the Exchange on the next
            // retry
            exchange.setProperty(Exchange.ROUTE_STOP, Boolean.TRUE);
        }

        // Set the ExchangeMessage status to canceled
        final Optional<ExchangeMessage> optionalExchangeMessage = this.messageService.findOneByExchangeId(exchangeId);

        if (optionalExchangeMessage.isPresent()) {
            this.messageService.setProcessingStatus(optionalExchangeMessage.get(), ProcessingStatus.CANCELLED, null);
        }
    }
}

我不得不承认，我很惊讶地看到该消息重新出现，因为它应该已从队列中删除。我们哪里做错了？

Answer 1

尝试在取消时将 InterruptedException 设置为异常 交换也是如此，这应该由重新交付错误检测到 处理程序和爆发，例如

exchange.setException(new InterruptedException("Cancel Exchange"));
exchange.setProperty(Exchange.ROUTE_STOP, Boolean.TRUE);