Apache Camel-失败的FTP组件

Question

我做了一个小写的骆驼来消耗一个ftp服务器。

Caused by: [org.apache.camel.component.file.GenericFileOperationFailedException - File operation failed: 150 Opening ASCII mode data connection for 2386442.XML(3895 bytes).
Accept timed out. Code: 150]
org.apache.camel.component.file.GenericFileOperationFailedException: File operation failed: 150 Opening ASCII mode data connection for 2386442.XML(3895 bytes).
Accept timed out. Code: 150
    at org.apache.camel.component.file.remote.FtpOperations.retrieveFileToStreamInBody(FtpOperations.java:336)
    at org.apache.camel.component.file.remote.FtpOperations.retrieveFile(FtpOperations.java:297)
    at org.apache.camel.component.file.GenericFileConsumer.processExchange(GenericFileConsumer.java:333)
    at org.apache.camel.component.file.remote.RemoteFileConsumer.processExchange(RemoteFileConsumer.java:94)
    at org.apache.camel.component.file.GenericFileConsumer.processBatch(GenericFileConsumer.java:175)
    at org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:136)
    at org.apache.camel.impl.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:140)
    at org.apache.camel.impl.ScheduledPollConsumer.run(ScheduledPollConsumer.java:92)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:441)
    at java.util.concurrent.FutureTask$Sync.innerRunAndReset(FutureTask.java:317)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:150)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$101(ScheduledThreadPoolExecutor.java:98)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.runPeriodic(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:204)
    at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
    at java.lang.Thread.run(Thread.java:662)
Caused by: java.net.SocketTimeoutException: Accept timed out
    at java.net.PlainSocketImpl.socketAccept(Native Method)
    at java.net.PlainSocketImpl.accept(PlainSocketImpl.java:408)
    at java.net.ServerSocket.implAccept(ServerSocket.java:462)
    at java.net.ServerSocket.accept(ServerSocket.java:430)
    at org.apache.commons.net.ftp.FTPClient._openDataConnection_(FTPClient.java:560)
    at org.apache.commons.net.ftp.FTPClient.retrieveFile(FTPClient.java:1442)
    at org.apache.camel.component.file.remote.FtpOperations.retrieveFileToStreamInBody(FtpOperations.java:328)
    ... 16 more
Caused by: [org.apache.camel.component.file.GenericFileOperationFailedException - Cannot retrieve file: GenericFile[2386448.XML] from: Endpoint[ftp://1.1.1.1?delay=15000&delete=true&disconnect=true&exclude=((?i).*pdf$)&password=******&username=user]
org.apache.camel.component.file.GenericFileOperationFailedException: Cannot retrieve file: GenericFile[2386448.XML] from:   Endpoint[ftp://1.1.1.1?delay=15000&delete=true&disconnect=true&exclude=((?i).*pdf$)&password=******&username=user]
   at org.apache.camel.component.file.GenericFileConsumer.processExchange(GenericFileConsumer.java:338)
  at org.apache.camel.component.file.remote.RemoteFileConsumer.processExchange(RemoteFileConsumer.java:94)
    at org.apache.camel.component.file.GenericFileConsumer.processBatch(GenericFileConsumer.java:175)
    at org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:136)
    at org.apache.camel.impl.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:140)
    at org.apache.camel.impl.ScheduledPollConsumer.run(ScheduledPollConsumer.java:92)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:441)
    at java.util.concurrent.FutureTask$Sync.innerRunAndReset(FutureTask.java:317)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:150)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$101(ScheduledThreadPoolExecutor.java:98)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.runPeriodic(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:204)
    at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
    at java.lang.Thread.run(Thread.java:662)

这是我使用Java DSL进行的路由：

    // XML Predicate
    // only allows names without spaces
    Predicate xmlPredicate = header(RssUtils.CAMEL_FILE_NAME).regex(
            "([\\S]+(\\.(?i)(xml))$)");
    // Images Predicate
    // only allows names without spaces
    Predicate imgPredicate = header(RssUtils.CAMEL_FILE_NAME).regex(
            "([\\S]+(\\.(?i)(jpg|png|gif))$)");

    onException(SchemaValidationException.class).to(
            "file://" + props.getProperty(RssUtils.ROOT_DIR)
                    + "/errors/SchemaValidationException");

    onException(GenericFileOperationFailedException.class).to(
            "file://" + props.getProperty(RssUtils.ROOT_DIR)
                    + "/errors/GenericFileExceptions");

    from(
            "ftp://"
                    + props.getProperty(RssUtils.FTP_URL)
                    + "?username="
                    + props.getProperty(RssUtils.FTP_USER)
                    + "&password="
                    + props.getProperty(RssUtils.FTP_PWD)
                    + "&disconnect=true&delete=true&exclude=((?i).*pdf$)&delay="
                    + props.getProperty(RssUtils.FTP_DELAY))
            .choice()
            .when(xmlPredicate)
            .to("jms:xmlQueue")
    .to("jms:archiveQueue")
            .when(imgPredicate)
            .to("file://" + props.getProperty(RssUtils.ROOT_DIR) + "/img")
            .otherwise()
            .to("file://" + props.getProperty(RssUtils.ROOT_DIR)
                    + "/errors/other");

    from("jms:xmlQueue").to("validator:FtpXmlValidator.xsd")
            .to("xslt://XmlToRssConverter.xsl")
            .process(rssFeedProcessor)
            .to("file://" + props.getProperty(RssUtils.ROOT_DIR) + "/rss/");

from("jms:archiveQueue")
    .to("file://" + props.getProperty(RssUtils.ROOT_DIR) + "/archive/");

我能做些什么来避免这种af行为吗？它真的很难测试，所以我希望有人发现我的代码中的一个缺陷。我已经找了很长一段时间，但我没有找到任何坚实的东西。也许我可以用什么方法来调试这个问题？

也许有几件事我发现有人可以说：

• 使用OnException时使用已处理(true)
• 可以设置使用者的最大批处理大小吗？（我可以为此使用油门吗？）
• 最后使用显式try catch，因为我使用的是Java DSL

如果我说错了别开枪，我只是学骆驼。因此，如果有人对上面的代码有建议，我将不胜感激！

提前多谢了！

Answer 1

这里是一个FTP问题，它发生在Apache Camel中的事实在很大程度上是无关的。

炸弹最能说明问题的部分是：

在java.net.PlainSockeTimpl.SocketAccept（本机方法）在java.net.PlainSockeTimpl.Accept（PlainSockeTimpl.java:408)在java.net.ServerSocket.ImplAccept（ServerSocket.java:462)在java.net.ServerSocket.Accept（ServerSocket.java:430)在org.apache.commons.net.ftp.ftpclient.OpenDataConnection（ftpclient.java:560)

org.apache.commons.net.FTP.ftpclient的openDataConnection方法支持主动模式FTP-被动模式只是使用与命令相同的端口，因此不需要单独的端口连接。

尝试切换到被动模式（使用Apache Camel时，passiveMode=true）。