如何解决RedHat Fuse+Camel+CXF适配器的Exchange超时问题

我使用基于Java DSL的RedHat Fuse适配器，结合Apache Camel与CXF处理大体积SOAP入站消息时遇到超时问题：当处理耗时较长时，发送方会收到500内部服务器错误，但我方实际处理流程正常；若处理速度较快，发送方则收到200响应，一切正常。

日志显示该超时为30000ms，提示信息为The OUT message was not received within: 30000 millis，但我无法找到该超时的配置位置及修改方式。尝试过cxf.timeout=60000等属性但未生效，且Cxf.ConnectionTimeOut与Cxf.ReceiveTimeout属于HTTP层面的超时，并非此内部应用超时。

已尝试通过并行处理缩短耗时，但核心问题是超时时间过短，希望了解该超时的具体定义及对应的配置属性。

错误日志 #

26-09-24 07:52:12.575 [XNIO-1 task-1] WARN  o.a.cxf.phase.PhaseInterceptorChain - Application {http://example.namespace.com/wsdl/Service/V1}MyService#{http://example.namespace.com/wsdl/Service/V1}processRequest has thrown exception, unwinding now
org.apache.cxf.interceptor.Fault: The OUT message was not received within: 30000 millis. Exchange[91F3A9AAC0D6092-0000000000000000]
    at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.checkFailure(CxfConsumer.java:350)
    at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.setResponseBack(CxfConsumer.java:322)
    at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.asyncInvoke(CxfConsumer.java:212)
    at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.invoke(CxfConsumer.java:161)
    ...
Caused by: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 30000 millis. Exchange[91F3A9AAC0D6092-0000000000000000]
    at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.asyncInvoke(CxfConsumer.java:210)
    ... 105 common frames omitted

路由代码（Java DSL） #

package com.example.routes;

import com.example.logging.CustomLogger;
import com.example.beans.TransformData;
import com.example.beans.UUIDGenerator;
import com.example.configuration.ExceptionConnectionErrorResolver;
import com.example.webservices.WebServiceConfig;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.builder.ThreadPoolBuilder;
import org.apache.camel.CamelContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.concurrent.ExecutorService;

@Component
public class MyServiceRoute extends RouteBuilder {
    @Autowired
    CustomLogger customLogger;

    @Autowired
    WebServiceConfig webServiceConfig;

    @Autowired
    UUIDGenerator generator;

    @Autowired
    ExceptionConnectionErrorResolver exceptionConnectionErrorResolver;

    @Override
    public void configure() throws Exception {

        CamelContext context = getContext();

        ExecutorService executorService = new ThreadPoolBuilder(context)
                .poolSize(150)
                .maxPoolSize(150)
                .maxQueueSize(-1)
                .build("SplitThreadPool");

        from("cxf:bean:processRequestSoapEndpoint?exchangePattern=InOnly").routeId("myServiceRoute")

        .onException(org.apache.camel.support.processor.validation.SchemaValidationException.class,org.xml.sax.SAXParseException.class,org.apache.camel.ValidationException.class).id("onTransformationException")
                .handled(true)
                .bean(customLogger, "logCode(${exchangeProperty.BreadCrumb}, CODE-992, Validation of message failed. ${exception.message})").id("logCode992")
                .process(exchange -> {
                    Exception exception = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class);
                    exchange.getContext().createProducerTemplate()
                            .sendBody("log:error?showAll=true&multiline=true",
                                    "CODE-992 Validation Error: " + exception.getMessage() + "\nStack Trace:\n" + getStackTrace(exception));
                })
        .end()

        .onException(javax.xml.transform.TransformerException.class).id("OnValidationException")
                .handled(true)
                .bean(customLogger, "logCode(${exchangeProperty.BreadCrumb}, CODE-032)").id("logCode032")
        .end()

        .onException(org.apache.cxf.interceptor.Fault.class,org.apache.camel.http.base.HttpOperationFailedException.class).id("onEndpointException")
                .handled(true)
                .bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-043)").id("logCode043")
                .bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-996,Error: ${exception.message})").id("logCode996")
        .end()

        .onException(java.net.ConnectException.class,org.apache.http.conn.HttpHostConnectException.class).id("onConnectionEndpoint")
                .handled(true)
                .maximumRedeliveries(exceptionConnectionErrorResolver.getMaxRedeliveries())
                .redeliveryDelay(exceptionConnectionErrorResolver.getRedeliveryDelay())
                .retryAttemptedLogLevel(LoggingLevel.WARN)
                .bean(customLogger, String.format("logCode(${exchangeProperty.BreadCrumb},CODE-042,%s)",webServiceConfig.processRequestSoapEndpointBackend().getAddress())).id("logCode042")
                .bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-996,Error: ${exception.message})").id("logCode996")
        .end()

        .onException(java.lang.Exception.class).id("onGeneralException")
                .handled(true)
                .bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-000)").id("logCode000")
                .bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-996,Error: ${exception.message})").id("logUnexpectedException")
        .end()

                .bean(customLogger, "logStart").id("logStart")
                .bean("PropertyGenerator").id("setProperties")
                .setProperty("messageHeaderProperty", xpath("/ns:processRequestHeader", XpathNameSpaces.namespaces)).id("setMessageHeaderProperty")
                .setProperty("voorloopRecordProperty", xpath("/ns:processRequestBody/ns:businessData/ns:record", XpathNameSpaces.namespaces)).id("setVoorloopRecordProperty")
                .to("validator:classpath:services/ExampleService/xsd/BM/V1/ExampleMessage.xsd").id("validateExampleMessage")
                .bean(customLogger, "logCode(${exchangeProperty.BreadCrumb}, CODE-092)").id("logCode092")
                .split(xpath("/ns:processRequestBody/ns:businessData/ns:record/ns:person", XpathNameSpaces.namespaces)).parallelProcessing().executorService(executorService).stopOnException().id("splitPerPerson")
                    .bean(TransformData.class).id("setSplittedBody")
                    .process(generator).id("setUUIDProp")
                    .setProperty("timestampProperty").simple("${date:now:yyyy-MM-dd'T'HH:mm:ss.SSS'Z'}").id("setTimestamp")
                    .to("xslt:classpath:/transformations/ExampleTransform.xsl").id("transformExample")
                    .to("validator:classpath:services/ExampleService/xsd/BAS/V1/ExampleValidation.xsd").id("validateExampleValidation")
                    .removeHeaders("*", "BreadCrumb").id("removeHeaders")
                    .to("cxf:bean:processRequestSoapEndpointBackend").id("toSoapEndpoint")
                    .bean(customLogger, String.format("logCode(${exchangeProperty.BreadCrumb}, CODE-041,%s)",webServiceConfig.processRequestSoapEndpointBackend().getAddress())).id("logCode041")
                .bean(customLogger, "logEnd").id("logCodeEnd");
    }

        private String getStackTrace(Exception exception) {
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            exception.printStackTrace(pw);
            return sw.toString();
        }
}

解决方案 #

问题根源 #

这个超时是Camel CXF消费者的异步调用超时，对应Camel CXF组件内部的asyncInvokeTimeout参数，默认值为30000ms（30秒）。它控制CXF消费者等待Camel路由处理完成并返回响应信号的最长时间，和HTTP层面的连接/接收超时不属于同一范畴。

修改配置的两种方式 #

方式1：直接在端点URL中添加参数

修改路由的from端点，追加asyncInvokeTimeout参数设置超时时间（示例设置为60秒）：

from("cxf:bean:processRequestSoapEndpoint?exchangePattern=InOnly&asyncInvokeTimeout=60000").routeId("myServiceRoute")

方式2：通过CXF端点Bean配置

如果你的processRequestSoapEndpoint是在Spring配置类中定义的Bean，可通过CxfEndpoint的API设置超时：

@Bean
public CxfEndpoint processRequestSoapEndpoint() {
    CxfEndpoint endpoint = new CxfEndpoint();
    // 其他原有配置（如地址、服务类等）
    endpoint.setAsyncInvokeTimeout(60000); // 单位：毫秒，根据业务需求调整
    return endpoint;
}

补充说明 #

• 即便使用InOnly交换模式，CXF消费者仍会等待路由处理完成的信号，这个超时就是控制该等待过程的最长时间。
• 设置超时时间时，需结合实际业务处理耗时合理调整，同时确保下游系统的超时配置与之匹配，避免出现其他层面的超时冲突。

内容的提问来源于stack exchange，提问作者Gerben Spoorwegen