java - Apache Camel 无限循环出错

Question

我在 Apache Camel (2.15.2) 中遇到无限循环问题。下面显示的路线在正常条件下工作正常。但是，在极少数情况下，会创建无限循环，从而消耗所有 CPU。我不确定是什么原因导致此问题，但我有一种感觉，这与在路由上生成消息的远程应用程序关闭时有关。

配置路由的代码(使用RouteBuilder):

@Override
public void configure() throws Exception {
    errorHandler(deadLetterChannel(
        "file:messages/myRoute?fileName=${date:now:yyyyMMdd-HHmm-ssSSS}.xml")
        .useOriginalMessage()
        .maximumRedeliveries(10)
        .delayPattern("1:5000");
    onException(java.net.ConnectException.class).handled(true);

    fromF("myEndpoint")
        .routeId("myRoute")
        .bean(messageHandler, "process");
}

处理消息的 bean:

public void process(final String message) {
    final MyMessage message = (MyMessage) marshaller.getXStream().fromXML(message);
    ...some processing...
    producer.sendBody("anotherEndpoint", marshaller.getXStream().toXML(message));
}

我做了几个线程转储，它们都显示了与此类似的内容:

at com.thoughtworks.xstream.core.DefaultConverterLookup.lookupConverterForType(DefaultConverterLookup.java:56)
at com.thoughtworks.xstream.core.TreeUnmarshaller.convertAnother(TreeUnmarshaller.java:56)
at com.thoughtworks.xstream.core.TreeUnmarshaller.convertAnother(TreeUnmarshaller.java:50)
at com.thoughtworks.xstream.core.TreeUnmarshaller.start(TreeUnmarshaller.java:134)
at com.thoughtworks.xstream.core.AbstractTreeMarshallingStrategy.unmarshal(AbstractTreeMarshallingStrategy.java:32)
at com.thoughtworks.xstream.XStream.unmarshal(XStream.java:1206)
at com.thoughtworks.xstream.XStream.unmarshal(XStream.java:1190)
at com.thoughtworks.xstream.XStream.fromXML(XStream.java:1061)
at com.thoughtworks.xstream.XStream.fromXML(XStream.java:1052)
at com.example.message.MessageHandler.process(MessageHandler.java:58)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.apache.camel.component.bean.MethodInfo.invoke(MethodInfo.java:408)
at org.apache.camel.component.bean.MethodInfo$1.doProceed(MethodInfo.java:279)
at org.apache.camel.component.bean.MethodInfo$1.proceed(MethodInfo.java:252)
at org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:171)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:109)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:87)
at org.apache.camel.component.jms.EndpointMessageListener.onMessage(EndpointMessageListener.java:103)
at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:746)
at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:684)
at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:651)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:315)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:253)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1164)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1156)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1053)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)

也许我的 errorHandler 和/或 onException 不正确？谢谢!

Answer 1

无限循环可能与底层 x-stream 库中的并发问题有关。请参阅https://github.com/x-stream/xstream/issues/2以及相关问题。