error-handling - 在 Apache Camel 中重新交付用尽后如何调用 Web 服务？

Question

我没有找到促进解决此问题的企业集成模式或方法:

重新发送尝试用尽后，我需要将 Web 服务请求发送回原始源，以通知发送者发送失败。

在用尽所有重新传递尝试后，我应该将消息移动到死信队列吗？然后创建一个新的消费者来监听那个 DL 队列？我的每个源消息队列都需要一个唯一的死信队列吗？我是否应该在将消息头移到死信队列之前添加消息头，注意源队列？如果所有消息都进入一个死信队列，我的消费者应该如何知道将 Web 服务请求发送到哪里？

你能指点我一本书、博客文章或文章吗？规定的方法是什么？

我正在使用一个非常旧版本的 Fuse ESB，但我希望 ServiceMix 中的解决方案同样适用。

或者，我要求的是反模式或代码气味。请指教。

Answer 1

如果您是 Camel 新手，并且真的想深入了解它，我会推荐 Camel in Action, a book by Claus Ibsen .有a second edition in the works ，已经完成了 19 章中的 14 章，因此您也可以试一试。

如果这有点太多，在线文档还不错，您可以从中找到基础知识。对于错误处理，我建议从 general error handling page 开始。然后转到 error handler docs和 exception policy documentation .

一般情况下，dead letter channel是要走的路 - 重试后 Camel 会自动发送到 DLC，你只需要自己定义 DLC。顾名思义，它是一个 channel ，实际上并不需要是一个队列——您可以写入文件、调用 Web 服务、将消息提交到消息队列或只写入日志，这完全取决于您。

// error-handler DLC, will send to HTTP endpoint when retries are exhausted
errorHandler(deadLetterChannel("http4://my.webservice.hos/path")
    .useOriginalMessage()
    .maximumRedeliveries(3)
    .redeliveryDelay(5000))

// exception-clause DLC, will send to HTTP endpoint when retries are exhausted
onException(NetworkException.class)
    .handled(true)
    .maximumRedeliveries(5)
    .backOffMultiplier(3)
    .redeliveryDelay(15000)
    .to("http4://my.webservice.hos/otherpath");

我自己一直更喜欢拥有一个消息队列，然后从那里消费以进行任何其他恢复或报告。我通常包括失败细节，比如交换 ID 和路由 ID、消息头、错误消息，有时甚至是堆栈跟踪。正如您可以想象的那样，生成的消息增长了很多，但它极大地简化了故障排除和调试，尤其是在您拥有大量组件和服务的环境中。这是来自我的一个项目的示例 DLC 消息:

public class DeadLetterChannelMessage {
  private String timestamp = Times.nowInUtc().toString();
  private String exchangeId;
  private String originalMessageBody;
  private Map<String, Object> headers;
  private String fromRouteId;
  private String errorMessage;
  private String stackTrace;

  @RequiredByThirdPartyFramework("jackson")
  private DeadLetterChannelMessage() {
  }

  @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
  public DeadLetterChannelMessage(Exchange e) {
    exchangeId = e.getExchangeId();
    originalMessageBody = e.getIn().getBody(String.class);
    headers = Collections.unmodifiableMap(e.getIn().getHeaders());
    fromRouteId = e.getFromRouteId();

    Optional.ofNullable(e.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class))
      .ifPresent(throwable -> {
        errorMessage = throwable.getMessage();
        stackTrace = ExceptionUtils.getStackTrace(throwable);
      });
  }

  // getters
}

从死信队列中消费时，路由 ID 可以告诉您故障源自何处，因此您可以实现特定于处理来自那里的错误的路由:

// general DLC handling route
from("{{your.dlc.uri}}")
    .routeId(ID_REPROCESSABLE_DLC_ROUTE)
    .removeHeaders(Headers.ALL)
    .unmarshal().json(JsonLibrary.Jackson, DeadLetterChannelMessage.class)
    .toD("direct:reprocess_${body.fromRouteId}"); // error handling route

// handle errors from `myRouteId`
from("direct:reprocess_myRouteId")  
    .log("Error: ${body.errorMessage} for ${body.originalMessageBody}"); 
    // you'll probably do something better here, e.g.
    // .convertBodyTo(WebServiceErrorReport.class) // requires a converter
    // .process(e -> { //do some pre-processing, like setting headers/properties })
    // .toD("http4://web-service-uri/path"); // send to web-service


// for routes that have no DLC handling supplied
onException(DirectConsumerNotAvailableException.class)
     .handled(true)
     .useOriginalMessage()
     .removeHeaders(Headers.ALL)
     .to({{my.unreprocessable.dlc}}); // errors that cannot be recovered from