Commons Net FTPClient无限期挂起Mule

Question

我遇到了Mule ESB FTP Transport的问题：当轮询时，运行客户端的线程将无限期地挂起而不会引发错误。这会导致FTP轮询完全停止。 Mule使用Apache Commons Net FTPClient。

进一步了解代码，我认为这是由于FTPClient的SocketTimeout没有被设置，有时在从FTPClient的套接字读取行时导致无限悬挂。

当问题发生时，我们可以清楚地看到用jstack检索到的堆栈中的问题。 __getReply（）函数似乎是更直接的问题。

这一个挂在连接（）创建一个新FTPClient时称：

receiver.172 prio=10 tid=0x00007f23e43c8800 nid=0x2d5 runnable [0x00007f24c32f1000] 
    java.lang.Thread.State: RUNNABLE 
    at java.net.SocketInputStream.socketRead0(Native Method) 
    at java.net.SocketInputStream.read(SocketInputStream.java:152) 
    at java.net.SocketInputStream.read(SocketInputStream.java:122) 
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:283) 
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:325) 
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:177) 
    - locked <0x00000007817a9578> (a java.io.InputStreamReader) 
    at java.io.InputStreamReader.read(InputStreamReader.java:184) 
    at java.io.BufferedReader.fill(BufferedReader.java:154) 
    at java.io.BufferedReader.readLine(BufferedReader.java:317) 
    - locked <0x00000007817a9578> (a java.io.InputStreamReader) 
    at java.io.BufferedReader.readLine(BufferedReader.java:382) 
    at org.apache.commons.net.ftp.FTP.__getReply(FTP.java:294) 
    at org.apache.commons.net.ftp.FTP._connectAction_(FTP.java:364) 
    at org.apache.commons.net.ftp.FTPClient._connectAction_(FTPClient.java:540) 
    at org.apache.commons.net.SocketClient.connect(SocketClient.java:178) 
    at org.mule.transport.ftp.FtpConnectionFactory.makeObject(FtpConnectionFactory.java:33) 
    at org.apache.commons.pool.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:1188) 
    at org.mule.transport.ftp.FtpConnector.getFtp(FtpConnector.java:172) 
    at org.mule.transport.ftp.FtpConnector.createFtpClient(FtpConnector.java:637) 
    at org.mule.transport.ftp.FtpMessageReceiver.listFiles(FtpMessageReceiver.java:134) 
    at org.mule.transport.ftp.FtpMessageReceiver.poll(FtpMessageReceiver.java:94) 
    at org.mule.transport.AbstractPollingMessageReceiver.performPoll(AbstractPollingMessageReceiver.java:216) 
    at org.mule.transport.PollingReceiverWorker.poll(PollingReceiverWorker.java:80) 
    at org.mule.transport.PollingReceiverWorker.run(PollingReceiverWorker.java:49) 
    at org.mule.transport.TrackingWorkManager$TrackeableWork.run(TrackingWorkManager.java:267) 
    at org.mule.work.WorkerContext.run(WorkerContext.java:286) 
    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) 

    Locked ownable synchronizers: 
    - <0x00000007817a3540> (a java.util.concurrent.ThreadPoolExecutor$Worker) 

而另挂在PASV（）使用listFiles时调用（）：

receiver.137" prio=10 tid=0x00007f23e433b000 nid=0x7c06 runnable [0x00007f24c2fee000] 
    java.lang.Thread.State: RUNNABLE 
    at java.net.SocketInputStream.socketRead0(Native Method) 
    at java.net.SocketInputStream.read(SocketInputStream.java:152) 
    at java.net.SocketInputStream.read(SocketInputStream.java:122) 
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:283) 
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:325) 
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:177) 
    - locked <0x0000000788847ed0> (a java.io.InputStreamReader) 
    at java.io.InputStreamReader.read(InputStreamReader.java:184) 
    at java.io.BufferedReader.fill(BufferedReader.java:154) 
    at java.io.BufferedReader.readLine(BufferedReader.java:317) 
    - locked <0x0000000788847ed0> (a java.io.InputStreamReader) 
    at java.io.BufferedReader.readLine(BufferedReader.java:382) 
    at org.apache.commons.net.ftp.FTP.__getReply(FTP.java:294) 
    at org.apache.commons.net.ftp.FTP.sendCommand(FTP.java:490) 
    at org.apache.commons.net.ftp.FTP.sendCommand(FTP.java:534) 
    at org.apache.commons.net.ftp.FTP.sendCommand(FTP.java:583) 
    at org.apache.commons.net.ftp.FTP.pasv(FTP.java:882) 
    at org.apache.commons.net.ftp.FTPClient._openDataConnection_(FTPClient.java:497) 
    at org.apache.commons.net.ftp.FTPClient.initiateListParsing(FTPClient.java:2296) 
    at org.apache.commons.net.ftp.FTPClient.initiateListParsing(FTPClient.java:2269) 
    at org.apache.commons.net.ftp.FTPClient.initiateListParsing(FTPClient.java:2189) 
    at org.apache.commons.net.ftp.FTPClient.initiateListParsing(FTPClient.java:2132) 
    at org.mule.transport.ftp.FtpMessageReceiver.listFiles(FtpMessageReceiver.java:135) 
    at org.mule.transport.ftp.FtpMessageReceiver.poll(FtpMessageReceiver.java:94) 
    at org.mule.transport.AbstractPollingMessageReceiver.performPoll(AbstractPollingMessageReceiver.java:216) 
    at org.mule.transport.PollingReceiverWorker.poll(PollingReceiverWorker.java:80) 
    at org.mule.transport.PollingReceiverWorker.run(PollingReceiverWorker.java:49) 
    at org.mule.transport.TrackingWorkManager$TrackeableWork.run(TrackingWorkManager.java:267) 
    at org.mule.work.WorkerContext.run(WorkerContext.java:286) 
    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) 

    Locked ownable synchronizers: 
    - <0x0000000788832180> (a java.util.concurrent.ThreadPoolExecutor$Worker) 

我觉得问题是由Mule默认FtpConnectionFactory中使用默认FTPClient构造函数（扩展SocketClient）引起的。

注setConnectTimeout（）值似乎调用socket.connect（）的时候才使用，但使用相同的插座上的其他操作忽略：

protected FTPClient createFtpClient() 
    { 
     FTPClient ftpClient = new FTPClient(); 
     ftpClient.setConnectTimeout(connectionTimeout); 

     return ftpClient; 
    } 

它使用FTPClient（）构造函数，本身使用具有0超时的SocketClient，在创建套接字时定义。

public SocketClient() 
    { 
     ... 
     _timeout_ = 0; 
     ... 
    } 

然后我们调用connec（），它调用_ connectAction（）_。

在SocketClient：

protected void _connectAction_() throws IOException 
    { 
     ... 
     _socket_.setSoTimeout(_timeout_); 
     ... 
    } 

在FTP，一个新的阅读器实例化以后，我们永远插座：

protected _connectAction_(){ 

    ... 
_controlInput_ = 
     new BufferedReader(new InputStreamReader(_socket_.getInputStream(), 
                getControlEncoding())); 
    ... 
} 

然后调用__getReply（）函数时，我们使用这个阅读器，与 - 永久插座：

private void __getReply() throws IOException 
    { 
     ... 
     String line = _controlInput_.readLine(); 
     ... 
    } 

对不起，很长的文章，但我认为这需要正确的解释秒。解决方案可能是在connect（）之后调用setSoTimeout（）来定义套接字超时。

有一个默认超时似乎不是一个可接受的解决方案，因为每个用户可能有不同的需求，并且在任何情况下默认不适合。 https://issues.apache.org/jira/browse/NET-35

最后，提出了2个问题：

1. 这似乎是我的错误，因为它会完全停止FTP轮询不给错误。你怎么看？
2. 什么可以是一种简单的方法来避免这种情况？用自定义的FtpConnectionFactory调用setSoTimeout（）？我在某处丢失配置或参数吗？

提前致谢。

编辑：我使用Mule CE Standalone 3.5.0，它似乎使用Apache Commons Net 2.0。但在代码中看来，Mule CE Standalone 3.7与Commons Net 2.2似乎并不相同。下面是源代码涉及：

https://github.com/mulesoft/mule/blob/mule-3.5.x/transports/ftp/src/main/java/org/mule/transport/ftp/FtpConnectionFactory.java

http://grepcode.com/file/repo1.maven.org/maven2/commons-net/commons-net/2.0/org/apache/commons/net/SocketClient.java

http://grepcode.com/file/repo1.maven.org/maven2/commons-net/commons-net/2.0/org/apache/commons/net/ftp/FTP.java

http://grepcode.com/file/repo1.maven.org/maven2/commons-net/commons-net/2.0/org/apache/commons/net/ftp/FTPClient.java

Answer 1

我在前面的案例中使用MockFtpServer重现了错误，并且我能够使用似乎解决该问题的FtpConnectionFactory。

public class SafeFtpConnectionFactory extends FtpConnectionFactory{ 

    //define a default timeout 
    public static int defaultTimeout = 60000; 
    public static synchronized int getDefaultTimeout() { 
     return defaultTimeout; 
    } 
    public static synchronized void setDefaultTimeout(int defaultTimeout) { 
     SafeFtpConnectionFactory.defaultTimeout = defaultTimeout; 
    } 

    public SafeFtpConnectionFactory(EndpointURI uri) { 
     super(uri); 
    } 

    @Override 
    protected FTPClient createFtpClient() { 
     FTPClient client = super.createFtpClient(); 

     //Define the default timeout here, which will be used by the socket by default, 
     //instead of the 0 timeout hanging indefinitely 
     client.setDefaultTimeout(getDefaultTimeout()); 

     return client; 
    } 
} 

然后将其连接到连接器我：

<ftp:connector name="archivingFtpConnector" doc:name="FTP" 
     pollingFrequency="${frequency}" 
     validateConnections="true" 
     connectionFactoryClass="my.comp.SafeFtpConnectionFactory"> 
    <reconnect frequency="${reconnection.frequency}" count="${reconnection.attempt}"/> 
</ftp:connector> 

使用该配置，一java.net.SocketTimeoutException将指定的超时之后被抛出，如：

java.net.SocketTimeoutException: Read timed out 
    at java.net.SocketInputStream.socketRead0(Native Method) 
    at java.net.SocketInputStream.read(SocketInputStream.java:152) 
    at java.net.SocketInputStream.read(SocketInputStream.java:122) 
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:283) 
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:325) 
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:177) 
    at java.io.InputStreamReader.read(InputStreamReader.java:184) 
    at java.io.BufferedReader.fill(BufferedReader.java:154) 
    at java.io.BufferedReader.readLine(BufferedReader.java:317) 
    at java.io.BufferedReader.readLine(BufferedReader.java:382) 
    at org.apache.commons.net.ftp.FTP.__getReply(FTP.java:294) 
    at org.apache.commons.net.ftp.FTP._connectAction_(FTP.java:364) 
    at org.apache.commons.net.ftp.FTPClient._connectAction_(FTPClient.java:540) 
    at org.apache.commons.net.SocketClient.connect(SocketClient.java:178) 
    at org.mule.transport.ftp.FtpConnectionFactory.makeObject(FtpConnectionFactory.java:33) 
    at org.apache.commons.pool.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:1188) 
    at org.mule.transport.ftp.FtpConnector.getFtp(FtpConnector.java:172) 
    at org.mule.transport.ftp.FtpConnector.createFtpClient(FtpConnector.java:637) 
    ... 

否则，尝试connect（）或pasv（）会无限期地挂起而没有服务器响应。我使用模拟FTP复制了这个确切的行为。

注：我用setDefaultTimeout（），因为它似乎是connect（）和connectAction（）中使用的变量（自SocketClient源）：

public abstract class SocketClient 
{ 
    ... 
    protected void _connectAction_() throws IOException 
    { 
     ... 
     _socket_.setSoTimeout(_timeout_); 
     ... 
    } 
    ... 
    public void setDefaultTimeout(int timeout) 
    { 
     _timeout_ = timeout; 
    } 
    ... 
} 

编辑：对于那些有兴趣谁，这里是模拟FTP的测试代码，用于重现从不应答的服务器。尽管如此，无限循环远非好习惯。它应该替换为像睡眠一样封闭的Test类，期望SocketTimeout异常并确保在给定超时后失败。

private static final int CONTROL_PORT = 2121; 

    public void startStubFtpServer(){ 
     FakeFtpServer fakeFtpServer = new FakeFtpServer(); 

     //define the command which should never be answered 
     fakeFtpServer.setCommandHandler(CommandNames.PASV, new EverlastingCommandHandler()); 
     //fakeFtpServer.setCommandHandler(CommandNames.CONNECT, new EverlastingConnectCommandHandler()); 
     //or any other command... 

     //server config 
     ... 

     //start server 
     fakeFtpServer.setServerControlPort(CONTROL_PORT); 
     fakeFtpServer.start(); 

     ... 

    } 

    //will cause any command received to never have an answer 
    public class EverlastingConnectCommandHandler extends org.mockftpserver.core.command.AbstractStaticReplyCommandHandler{ 
     @Override 
     protected void handleCommand(Command cmd, Session session, InvocationRecord rec) throws Exception { 
      while(true){ 
       try { 
        Thread.sleep(60000); 
       } catch (InterruptedException e) { 
        //TODO 
       } 
      } 
     } 

    } 
    public class EverlastingCommandHandler extends AbstractFakeCommandHandler { 
     @Override 
     protected void handle(Command cmd, Session session) { 
      while(true){ 
       try { 
        Thread.sleep(60000); 
       } catch (InterruptedException e) { 
        //TODO 
       } 
      } 
     } 
    }; 

Answer 2

在一个理想的世界超时不应该是必要的，但它看起来像你的情况下，是。

您的描述非常全面，您是否考虑过筹集bug？

解决方法我建议先在高级选项卡中使用“响应超时”。如果这不起作用，我会使用service override，从那里你应该能够覆盖接收器。