OpenHub框架进行的异步通信

在本系列的前一部分中，我们介绍了OpenHub框架。这部分显示了框架最强大的功能之一- 异步消息传递模型。

当源系统无法等待目标系统的响应时，将使用系统之间的异步通信。有以下几个原因：

源系统必须尽可能地响应 ，并且不受外部影响（通信缓慢，目标系统不稳定等）的影响
目标系统中的处理需要很多时间
异步通信对性能和流量产生积极影响

异步方案

当您决定异步通信时，您必须考虑可能的情况：

目标系统必须确认传入消息已成功保存 ，并准备进行进一步处理。是否应将异步过程的最终结果通知源系统？
如果异步处理失败怎么办？ 如果存在暂时性技术错误 （例如，与另一个系统的通信失败）， 请尝试几次，或者由于业务错误 （例如，输入数据无效）而停止进一步处理 。
在异步处理期间会调用其他系统- 如果对第一个系统的调用可以，但对第二个系统的调用失败，该怎么办？ 异步处理必须是幂等的，并跳过第一个成功的呼叫，然后仅重试第二个呼叫。
异步过程可能很棘手，因此最好将一个大进程（父进程）划分为较小的（子进程）进程 。如果处理了子进程，则父进程也将完成。
有时，您必须保证传入请求的顺序 （请求的到达顺序与发送的顺序不相同）并按照确切的顺序进行处理。
它是异步处理，您需要对其进行监视或在发生意外情况 （例如异步过程失败） 时自动得到通知 。
有时，您需要保存数据或异步过程的当前状态以使其成功完成，例如，第一次调用外部系统的结果就是第二次调用的输入。

当您开始考虑所有这些情况时，您会发现从头开始实施它并不简单。 OpenHub框架内置了对异步消息处理的支持。它易于使用，但同时又强大又灵活。并且也是可配置的，例如，该过程应再次运行多少次？在哪个时间间隔？

异步路由实现

使用OpenHub框架的路由实现有两个子路由：

一种用于处理传入消息（ RouteIn ）
一种用于异步过程实现（ RouteOut ）

/*** Route definition for asynchronous operation "translate" via web services.*/
@CamelConfiguration(value = AsyncTranslateWsRoute.ROUTE_BEAN)
public class AsyncTranslateWsRoute extends AbstractBasicRoute {static final String ROUTE_BEAN = "asyncTranslateWsRouteBean";private static final String OPERATION_NAME = "asyncTranslateWs";static final String ROUTE_ID_ASYNC_IN = getInRouteId(ServiceEnum.TRANSLATE, OPERATION_NAME);static final String ROUTE_ID_ASYNC_OUT = getOutRouteId(ServiceEnum.TRANSLATE, OPERATION_NAME);static final String URI_ASYNC_OUT = "direct:" + ROUTE_ID_ASYNC_OUT;@Overrideprotected void doConfigure() throws Exception {// asyncTranslate - input asynch messagecreateAsyncRouteIn();// asyncTranslate - process delivery (=asynchronous execution)createAsyncRouteOut();}/*** Route for asynchronous <strong>asyncTranslate</strong> input operation.* <p/>* Prerequisite: none* <p/>* Output: {@link AsyncTranslateResponse}*/private void createAsyncRouteIn() {Namespaces ns = new Namespaces("h", TranslateWebServiceConfig.TRANSLATE_SERVICE_NS);// note: mandatory parameters are set already in XSD, this validation is extraXPathValidator validator = new XPathValidator("/h:asyncTranslateRequest", ns, "h:inputText");AsynchRouteBuilder.newInstance(ServiceEnum.TRANSLATE, OPERATION_NAME,getInWsUri(new QName(TranslateWebServiceConfig.TRANSLATE_SERVICE_NS, "asyncTranslateRequest")),new AsynchResponseProcessor() {@Overrideprotected Object setCallbackResponse(CallbackResponse callbackResponse) {AsyncTranslateResponse res = new AsyncTranslateResponse();res.setConfirmAsyncTranslate(callbackResponse);return res;}}, jaxb(AsyncTranslateResponse.class)).withValidator(validator).build(this);}/*** Route for <strong>asyncTranslate</strong> operation - process delivery (=asynchronous execution).* Only input text is logged in this case.* <p/>* Prerequisite: none*/private void createAsyncRouteOut() {final String URI_LOG_INPUT_PARAMS = "direct:logInputParams";from(URI_ASYNC_OUT).routeId(ROUTE_ID_ASYNC_OUT)// xml -> AsyncTranslateRequest.unmarshal(jaxb(AsyncTranslateRequest.class)).to("extcall:message:" + URI_LOG_INPUT_PARAMS);from(URI_LOG_INPUT_PARAMS).validate(body().isInstanceOf(AsyncTranslateRequest.class)).log(LoggingLevel.DEBUG, "Asynchronous execution - input text '${body.inputText}' (lang: ${body.inputLang})");}
}

RouteIn使用AsynchRouteBuilder通过以下功能轻松配置：

定义哪个传入的Web Service请求应开始此路由
定义源系统的确认响应。输入路由成功执行后，将返回对源系统的同步响应。
定义验证器，该验证器检查传入请求中是否存在元素inputText

RouteOut定义了异步过程本身。输入请求（ AsyncTranslateRequest ）仅在这种情况下记录。

就这样。周围的一切都由OpenHub框架实现。

外部通话

您的路线实施通常会调用外部系统或其他路线。如果实现异步流程，则必须遵守幂等规则-流程的每个部分都可以被多次调用，并且必须确保所有调用的行为相同。有时外部系统/路由本身是幂等的，然后您可以根据需要多次调用它。如果没有，那么您必须在实现中对其进行控制。因此，我们将Camel组件设为extcall 。

上例中的组件excall确保即使整个异步过程运行了多次，使用URI_LOG_INPUT_PARAMS路由也会被精确地调用一次。