natty的异步通信框架_OpenHub框架进行的异步通信

natty的异步通信框架

在本系列的前一部分中，我们介绍了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通过以下功能轻松配置：