ftp连接池客户端

Posted 2021-02-03 sxpdy8571

package com.scenetec.isv.utils.ftp.core;

import com.scenetec.isv.utils.ftp.config.FtpClientProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.springframework.stereotype.Component;

/**
 * @author [email protected]
 * @description ftpclient客户端连接工厂类
 * @date 2018/12/19
 */
@Slf4j
@Component
public class FtpClientFactory extends BasePooledObjectFactory<FTPClient> {

   private FtpClientProperties config;

   public FtpClientFactory(FtpClientProperties config) {
      this.config = config;
   }

   @Override
   public FTPClient create() {

      FTPClient ftpClient = new FTPClient();
      ftpClient.setControlEncoding(config.getEncoding());
      if (config.getConnectTimeout() != null) {
         ftpClient.setConnectTimeout(config.getConnectTimeout());
      }

      try {
         ftpClient.connect(config.getHost(), config.getPort());
         int replyCode = ftpClient.getReplyCode();
         if (!FTPReply.isPositiveCompletion(replyCode)) {
            ftpClient.disconnect();
            log.warn("FTPServer refused connection,replyCode:{}", replyCode);
            return null;
         }

         if (!ftpClient.login(config.getUsername(), config.getPassword())) {
            log.warn("FTPClient login failed... username is {}; password: {}", config.getUsername(), config.getPassword());
         }

         ftpClient.setBufferSize(config.getBufferSize());
         ftpClient.setFileType(config.getTransferFileType());
         if (config.isPassiveMode()) {
            ftpClient.enterLocalPassiveMode();
         }

      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("Failed to create FTP connection");
      }

      return ftpClient;
   }

   void destroyObject(FTPClient client) {
      destroyObject(wrap(client));
   }

   /**
    * 用PooledObject封装对象放入池中
    * @param ftpClient ftp客户端
    * @return 默认池对象
    */
   @Override
   public PooledObject<FTPClient> wrap(FTPClient ftpClient) {
      return new DefaultPooledObject<>(ftpClient);
   }

   /**
    * 销毁FtpClient对象
    * @param ftpPooled ftp池对象
    */
   @Override
   public void destroyObject(PooledObject<FTPClient> ftpPooled) {
      if (ftpPooled == null) {
         return;
      }

      FTPClient ftpClient = ftpPooled.getObject();

      try {
         if (ftpClient.isConnected()) {
            ftpClient.logout();
         }
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("Failure to destroy FTP connection pool.");
      } finally {
         try {
            ftpClient.disconnect();
         } catch (Exception ex) {
            ex.printStackTrace();
            log.error("Failed to close FTP connection pool.");
         }
      }
   }

   /**
    * 验证FtpClient对象
    * @param ftpPooled ftp池对象
    * @return 发送一个NOOP命令到FTP服务器，如果成功返回true，否则为false。
    */
   @Override
   public boolean validateObject(PooledObject<FTPClient> ftpPooled) {
      try {
         FTPClient ftpClient = ftpPooled.getObject();
         if (ftpClient == null) {
            return false;
         }
         if (!ftpClient.isConnected()) {
            return false;
         }
         return ftpClient.sendNoOp();
      } catch (Exception ex) {
         ex.printStackTrace();
      }
      return false;
   }

}

package com.scenetec.isv.utils.ftp.core;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.pool2.BaseObjectPool;
import org.springframework.util.ObjectUtils;

import java.util.NoSuchElementException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author [email protected]
 * @description ftp客户端连接池类
 * @date 2018/12/19
 */
@Slf4j
@Deprecated
public class FtpClientPool extends BaseObjectPool<FTPClient> {

   private static final int DEFAULT_POOL_SIZE = 5;
   private final BlockingQueue<FTPClient> pool;
   private final FtpClientFactory factory;

   /**
    * 初始化连接池，需要注入一个工厂来提供FTPClient实例
    * @param factory
    */
   public FtpClientPool(FtpClientFactory factory) {
      this(DEFAULT_POOL_SIZE, factory);
   }

   public FtpClientPool(int poolSize, FtpClientFactory factory) {
      this.factory = factory;
      this.pool = new ArrayBlockingQueue<>(poolSize * 2);
      initPool(poolSize);
   }

   /**
    * 初始化连接池，需要注入一个工厂来提供FTPClient实例
    * @param maxPoolSize
    */
   private void initPool(int maxPoolSize) {
      try {
         for (int i = 0; i < maxPoolSize; i++) {
            addObject();
         }
      } catch (Exception ex) {
         ex.printStackTrace();
         throw new RuntimeException("Failed to initialize FTP thread pool.");
      }
   }

   /**
    * 客户端从池中借出一个对象
    * @return
    * @throws Exception
    * @throws NoSuchElementException
    * @throws IllegalStateException
    */
   @Override
   public FTPClient borrowObject() throws Exception, NoSuchElementException, IllegalStateException {
      FTPClient client = pool.take();
      if (ObjectUtils.isEmpty(client)) {
         // 创建新的连接
         client = factory.create();
         // 返换对象
         returnObject(client);
      }
      // 验证对象是否有效
      else if (!factory.validateObject(factory.wrap(client))) {
         // 对无效的对象进行处理
         invalidateObject(client);
         // 创建新的对象
         client = factory.create();
         // 将新的对象放入连接池
         returnObject(client);
      }
      // 返回ftp对象
      return client;
   }

   /**
    * 返还对象到连接池中
    * @param client
    * @throws Exception
    */
   @Override
   public void returnObject(FTPClient client) {
      try {
         long timeout = 3L;
         if (client != null) {
            if (client.isConnected()) {
               if (pool.size() < DEFAULT_POOL_SIZE) {
                  if (!pool.offer(client, timeout, TimeUnit.SECONDS)) {
                     factory.destroyObject(client);
                  }
               } else {
                  factory.destroyObject(client);
               }
            } else {
               factory.destroyObject(client);
            }
         }
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("Failed to return FTP connection object.");
      }
   }

   /**
    * 移除无效的对象
    * @param client
    * @throws Exception
    */
   @Override
   public void invalidateObject(FTPClient client) {
      try {
         // 移除无效对象
         pool.remove(client);
         // 注销对象
         factory.destroyObject(client);
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("Failed to remove invalid FTP object.");
      }
   }

   /**
    * 增加一个新的链接，超时失效
    * @throws Exception
    * @throws IllegalStateException
    * @throws UnsupportedOperationException
    */
   @Override
   public void addObject() throws Exception, IllegalStateException, UnsupportedOperationException {
      pool.offer(factory.create(), 3, TimeUnit.SECONDS);
   }

   @Override
   public void close() {
      try {
         while (pool.iterator().hasNext()) {
            FTPClient client = pool.take();
            factory.destroyObject(client);
         }
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("Failed to close FTP object.");
      }
   }
}

package com.scenetec.isv.utils.ftp.core;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.springframework.stereotype.Component;

import java.io.*;

/**
 * @author [email protected]
 * @description ftpClient模板上传下载文件类
 * @date 2018/12/20
 */
@Slf4j
@Component
public class FtpClientTemplate {

   private GenericObjectPool<FTPClient> ftpClientPool;

   public FtpClientTemplate(FtpClientFactory ftpClientFactory) {
      this.ftpClientPool = new GenericObjectPool<>(ftpClientFactory);
   }

   /**
    * 上传文件
    *
    * @param localPath 本地路径
    * @param remotePath 远程路径，必须包含文件名（"/"为当前ftp用户根路径）
    * @return 上传成功返回true， 否则返回false
    */
   public boolean uploadFile(String localPath, String remotePath) {
      if (StringUtils.isBlank(localPath)) {
         log.error("本地文件路径为空");
         return false;
      }
      return uploadFile(new File(localPath), remotePath);
   }

   /**
    * 上传文件
    *
    * @param localFile 本地文件
    * @param remotePath 远程文件，必须包含文件名
    * @return 上传成功返回true， 否则返回false
    */
   public boolean uploadFile(File localFile, String remotePath) {
      if (!localFile.exists()) {
         log.error("本地文件不存在");
         return false;
      }
      if (!localFile.isFile()) {
         log.error("上传类型不是文件");
      }
      if (StringUtils.isBlank(remotePath)) {
         remotePath = "/";
      }
      FileInputStream fis = null;
      BufferedInputStream bis = null;
      try {
         fis = new FileInputStream(localFile);
         bis = new BufferedInputStream(fis);
         // 上传
         return uploadFile(bis, remotePath);
      } catch (FileNotFoundException fex) {
         fex.printStackTrace();
         log.error("系统找不到指定的文件：{}", localFile);
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("上传文件异常。");
      } finally {
         try {
            if (bis != null) {
               bis.close();
            }
         } catch (Exception ex) {
            ex.printStackTrace();
         }
         try {
            if (fis != null) {
               fis.close();
            }
         } catch (Exception ex) {
            ex.printStackTrace();
         }
      }
      return false;
   }

   /**
    * 上传文件
    *
    * @param fileContent 文件内容
    * @param remotePath 远程文件，必须包含文件名
    * @return 上传成功返回true， 否则返回false
    */
   public boolean uploadFile(byte[] fileContent, String remotePath) {
      if (fileContent == null || fileContent.length <= 0) {
         log.error("上传文件内容为空。");
         return false;
      }
      InputStream is = null;
      try {
         is = new ByteArrayInputStream(fileContent);
         // 上传
         return uploadFile(is, remotePath);
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("上传文件异常。原因：【{}】", ex.getMessage());
      } finally {
         try {
            if (is != null) {
               is.close();
            }
         } catch (Exception ex) {
            ex.printStackTrace();
         }
      }
      return false;
   }

   /**
    * 上传文件
    *
    * @param inputStream 文件流
    * @param remotePath 远程文件，必须包含文件名
    * @return 上传成功返回true， 否则返回false
    */
   public boolean uploadFile(InputStream inputStream, String remotePath) {
      // 上传
      FTPClient client = null;
      try {
         // 获取远程文件路径
         String remoteFilePath = getRemoteFilePath(remotePath);
         // 获取远程文件名
         String remoteFileName = getRemoteFileName(remotePath);
         // 从池中获取对象
         client = getFtpClient();
         // 验证连接是否成功
         boolean bool = FTPReply.isPositiveCompletion(client.getReplyCode());
         if (!bool) {
            log.error("连接服务器失败，{}", client.getReplyString());
            return false;
         }
         // 切换工作路径
         bool = changeDirectory(client, remoteFilePath);
         if (!bool) {
            log.error("切换工作路径失败，{}", client.getReplyString());
            return false;
         }
         // 设置重试次数
         final int retryTime = 3;
         boolean retryResult = false;

         for (int i = 0; i <= retryTime; i++) {
            boolean success = client.storeFile(remoteFileName, inputStream);
            if (success) {
               log.info("文件【{}】上传成功。", remotePath);
               retryResult = true;
               break;
            } else {
               log.error("文件上传失败。{}", client.getReplyString());
            }
            log.warn("文件【{}】上传失败,重试上传...尝试{}次", remotePath, i);
         }

         return retryResult;
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("上传文件异常。");
      } finally {
         // 将对象放回池中
         if (client != null) {
            ftpClientPool.returnObject(client);
         }
      }

      return false;
   }

   /**
    * 下载文件
    * @param remotePath 远程文件，必须包含文件名
    * @param localPath 本地路径，必须包含文件名（全路径）
    * @return 下载成功返回true，否则返回false
    */
   public boolean downloadFile(String remotePath, String localPath) {
      if (StringUtils.isBlank(remotePath)) {
         remotePath = "/";
      }
      if (StringUtils.isBlank(localPath)) {
         log.error("本地文件路径为空");
         return false;
      }
      // 创建本地文件路径
      File localFile = new File(localPath);
      if (!localFile.exists()) {
         File parentFile = localFile.getParentFile();
         if (!parentFile.exists()) {
            boolean bool = parentFile.mkdirs();
            if (!bool) {
               log.error("创建本地路径失败");
               return false;
            }
         }
      }
      // 下载
      FTPClient client = null;
      OutputStream os = null;
      try {
         os = new FileOutputStream(localPath);
         // 获取远程文件路径
         String remoteFilePath = getRemoteFilePath(remotePath);
         // 获取远程文件名
         String remoteFileName = getRemoteFileName(remotePath);
         // 从池中获取对象
         client = getFtpClient();
         // 验证连接是否成功
         boolean bool = FTPReply.isPositiveCompletion(client.getReplyCode());
         if (!bool) {
            log.error("连接服务器失败，{}", client.getReplyString());
            return false;
         }
         // 切换工作路径
         bool = client.changeWorkingDirectory(remoteFilePath);
         if (!bool) {
            log.error("切换工作路径失败，{}", client.getReplyString());
            return false;
         }
         // 设置重试次数
         final int retryTime = 3;
         boolean retryResult = false;

         for (int i = 0; i <= retryTime; i++) {
            boolean success = client.retrieveFile(remoteFileName, os);
            if (success) {
               log.info("文件【{}】下载成功。", localPath);
               retryResult = true;
               break;
            } else {
               log.error("文件下载失败。 {}", client.getReplyString());
            }
            log.warn("文件【{}】下载失败，重试下载...尝试{}次", localPath, i);
         }
         // 返回结果
         return retryResult;
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("下载文件异常。");
      } finally {
         if (client != null) {
            ftpClientPool.returnObject(client);
         }
         try {
            if (os != null) {
               os.close();
            }
         } catch (Exception ex) {
            ex.printStackTrace();
         }
      }
      return false;
   }

   /**
    * 删除文件
    * @param remotePath 远程文件，必须包含文件名
    * @return 下载成功返回true，否则返回false
    */
   public boolean deleteFile(String remotePath) {
      FTPClient client = null;
      try {
         // 从池中获取对象
         client = getFtpClient();
         // 验证连接是否成功
         boolean bool = FTPReply.isPositiveCompletion(client.getReplyCode());
         if (!bool) {
            log.error("连接服务器失败，{}", client.getReplyString());
            return false;
         }
         // 删除文件
         bool = client.deleteFile(remotePath);
         if (!bool) {
            log.error("删除文件失败，{}", client.getReplyString());
         }
         return bool;
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("删除文件异常。");
      } finally {
         if (client != null) {
            ftpClientPool.returnObject(client);
         }
      }
      return false;
   }

   /**
    * 获取远程工作目录
    * @param remotePath 远程路径
    * @return 返回工作路径
    */
   private String getRemoteFilePath(String remotePath) {
      if (StringUtils.isNotBlank(remotePath)) {
         return remotePath.substring(0, remotePath.lastIndexOf("/") + 1);
      }
      return "/";
   }

   /**
    * 获取远程文件名
    * @param remotePath 远程路径
    * @return 返回文件名
    */
   private String getRemoteFileName(String remotePath) {
      if (StringUtils.isNotBlank(remotePath)) {
         return remotePath.substring(remotePath.lastIndexOf("/") + 1);
      }
      return "";
   }

   /**
    * 验证连接是否成功
    * @return 连接登录成功返回true，否则返回false
    */
   private FTPClient getFtpClient () {
      FTPClient client = null;
      try {
         while (true) {
            // 获取客户端
            client = ftpClientPool.borrowObject();
            // 验证客户端
            if (client == null) {
               continue;
            } else {
               if (!client.isConnected() || !FTPReply.isPositiveCompletion(client.getReplyCode())) {
                  ftpClientPool.invalidateObject(client);
               } else {
                  break;
               }
            }
         }
      } catch (Exception ex) {
         ex.printStackTrace();
      }
      return client;
   }

   /**
    * 切换工作目录
    * @param client ftp客户端
    * @param dir 工作目录
    * @return 切换成功返回true，否则返回false
    */
   private boolean changeDirectory(FTPClient client, String dir) {
      try {
         if (client == null || StringUtils.isBlank(dir)) {
            return false;
         }

         String fileBackslashSeparator = "\";
         String fileSeparator = "/";

         if (StringUtils.contains(dir, fileBackslashSeparator)) {
            dir = StringUtils.replaceAll(dir, fileBackslashSeparator, fileSeparator);
         }

         String[] dirArray = StringUtils.split(dir, fileSeparator);

         String tmp = "";
         for (String aDirArray : dirArray) {
            tmp += fileSeparator + aDirArray;
            if (!client.changeWorkingDirectory(tmp)) {
               // 创建工作目录
               client.makeDirectory(tmp);
               // 切换工作目录
               client.changeWorkingDirectory(tmp);
            }
         }
         return true;
      } catch (Exception ex) {
         ex.printStackTrace();
         log.error("切换工作目录失败。");
      }
      return false;
   }

}