快捷導(dǎo)航

Spring Boot整合FTPClient線程池的實(shí)現(xiàn)示例

更新時(shí)間：2018年12月23日 17:10:26 作者：hrabbits

這篇文章主要介紹了Spring Boot整合FTPClient線程池的實(shí)現(xiàn)示例，小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧

最近在寫一個(gè)FTP上傳工具，用到了Apache的FTPClient，但是每個(gè)線程頻繁的創(chuàng)建和銷毀FTPClient對(duì)象對(duì)服務(wù)器的壓力很大，因此，此處最好使用一個(gè)FTPClient連接池。仔細(xì)翻了一下Apache的api，發(fā)現(xiàn)它并沒(méi)有一個(gè)FTPClientPool的實(shí)現(xiàn)，所以，不得不自己寫一個(gè)FTPClientPool。下面就大體介紹一下開(kāi)發(fā)連接池的整個(gè)過(guò)程，供大家參考。

我們可以利用Apache提供的common-pool包來(lái)協(xié)助我們開(kāi)發(fā)連接池。而開(kāi)發(fā)一個(gè)簡(jiǎn)單的對(duì)象池，僅需要實(shí)現(xiàn)common-pool 包中的ObjectPool和PoolableObjectFactory兩個(gè)接口即可。

線程池的意義

為了減少頻繁創(chuàng)建、銷毀對(duì)象帶來(lái)的性能消耗，我們可以利用對(duì)象池的技術(shù)來(lái)實(shí)現(xiàn)對(duì)象的復(fù)用。對(duì)象池提供了一種機(jī)制，它可以管理對(duì)象池中對(duì)象的生命周期，提供了獲取和釋放對(duì)象的方法，可以讓客戶端很方便的使用對(duì)象池中的對(duì)象。

pom引入依賴

 <!-- FtpClient依賴包-->
    <dependency>
      <groupId>commons-net</groupId>
      <artifactId>commons-net</artifactId>
      <version>3.5</version>
    </dependency>

    <!-- 線程池-->
    <dependency>
      <groupId>commons-pool</groupId>
      <artifactId>commons-pool</artifactId>
      <version>1.6</version>
    </dependency>

    <dependency>
      <groupId>org.apache.commons</groupId>
      <artifactId>commons-pool2</artifactId>
      <version>2.0</version>
    </dependency>

創(chuàng)建ftp配置信息

在resources目錄下創(chuàng)建ftp.properties配置文件,目錄結(jié)構(gòu)如下：

添加如下的配置信息:

########### FTP用戶名稱 ###########
ftp.userName=hrabbit
########### FTP用戶密碼 ###########
ftp.passWord=123456
########### FTP主機(jī)IP ###########
ftp.host=127.0.0.1
########### FTP主機(jī)端口號(hào) ###########
ftp.port=21
########### 保存根路徑 ###########
ftp.baseUrl=/

創(chuàng)建FTPProperties.java配置文件

加載配置內(nèi)容到Spring中,配置信息基本延用我的就可以。

/**
 * FTP的配置信息
 * @Auther: hrabbit
 * @Date: 2018-12-03 2:06 PM
 * @Description:
 */
@Data
@Component
@PropertySource("classpath:ftp.properties")
@ConfigurationProperties(prefix = "ftp")
public class FTPProperties {

  private String username;

  private String password;

  private String host;

  private Integer port;

  private String baseUrl;

  private Integer passiveMode = FTP.BINARY_FILE_TYPE;

  private String encoding="UTF-8";

  private int clientTimeout=120000;

  private int bufferSize;

  private int transferFileType=FTP.BINARY_FILE_TYPE;

  private boolean renameUploaded;

  private int retryTime;
}

創(chuàng)建FTPClientPool線程池

/**
 * 自定義實(shí)現(xiàn)ftp連接池
 * @Auther: hrabbit
 * @Date: 2018-12-03 3:40 PM
 * @Description:
 */
@Slf4j
@SuppressWarnings("all")
public class FTPClientPool implements ObjectPool<FTPClient> {

  private static final int DEFAULT_POOL_SIZE = 10;

  public BlockingQueue<FTPClient> blockingQueue;

  private FTPClientFactory factory;

  public FTPClientPool(FTPClientFactory factory) throws Exception {
    this(DEFAULT_POOL_SIZE, factory);
  }

  public FTPClientPool(int poolSize, FTPClientFactory factory) throws Exception {
    this.factory = factory;
    this.blockingQueue = new ArrayBlockingQueue<FTPClient>(poolSize);
    initPool(poolSize);
  }

  /**
   * 初始化連接池
   * @param maxPoolSize
   *         最大連接數(shù)
   * @throws Exception
   */
  private void initPool(int maxPoolSize) throws Exception {
    int count = 0;
    while(count < maxPoolSize) {
      this.addObject();
      count++;
    }
  }

  /**
   * 從連接池中獲取對(duì)象
   */
  @Override
  public FTPClient borrowObject() throws Exception {
    FTPClient client = blockingQueue.take();
    if(client == null) {
      client = factory.makeObject();
    } else if(!factory.validateObject(client)) {
      invalidateObject(client);
      client = factory.makeObject();
    }
    return client;
  }

  /**
   * 返還一個(gè)對(duì)象(鏈接)
   */
  @Override
  public void returnObject(FTPClient client) throws Exception {
    if ((client != null) && !blockingQueue.offer(client,2,TimeUnit.MINUTES)) {
      try {
        factory.destroyObject(client);
      } catch (Exception e) {
        throw e;
      }
    }
  }

  /**
   * 移除無(wú)效的對(duì)象(FTP客戶端)
   */
  @Override
  public void invalidateObject(FTPClient client) throws Exception {
    blockingQueue.remove(client);
  }

  /**
   * 增加一個(gè)新的鏈接，超時(shí)失效
   */
  @Override
  public void addObject() throws Exception {
    blockingQueue.offer(factory.makeObject(), 2, TimeUnit.MINUTES);
  }

  /**
   * 重新連接
   */
  public FTPClient reconnect() throws Exception {
    return factory.makeObject();
  }

  /**
   * 獲取空閑鏈接數(shù)(這里暫不實(shí)現(xiàn))
   */
  @Override
  public int getNumIdle() {
    return blockingQueue.size();
  }

  /**
   * 獲取正在被使用的鏈接數(shù)
   */
  @Override
  public int getNumActive() {
    return DEFAULT_POOL_SIZE - getNumIdle();
  }

  @Override
  public void clear() throws Exception {

  }

  /**
   * 關(guān)閉連接池
   */
  @Override
  public void close() {
    try {
      while(blockingQueue.iterator().hasNext()) {
        FTPClient client = blockingQueue.take();
        factory.destroyObject(client);
      }
    } catch(Exception e) {
      log.error("close ftp client pool failed...{}", e);
    }
  }

  /**
   * 增加一個(gè)新的鏈接，超時(shí)失效
   */
  public void addObject(FTPClient ftpClient) throws Exception {
    blockingQueue.put(ftpClient);
  }
}

創(chuàng)建一個(gè)FTPClientFactory工廠類

創(chuàng)建FTPClientFactory實(shí)現(xiàn)PoolableObjectFactory的接口，F(xiàn)TPClient工廠類，通過(guò)FTPClient工廠提供FTPClient實(shí)例的創(chuàng)建和銷毀

/**
 * FTPClient 工廠
 * @Auther: hrabbit
 * @Date: 2018-12-03 3:41 PM
 * @Description:
 */
@Slf4j
@SuppressWarnings("all")
public class FTPClientFactory implements PoolableObjectFactory<FTPClient> {

  private FTPProperties ftpProperties;

  public FTPClientFactory(FTPProperties ftpProperties) {
    this.ftpProperties = ftpProperties;
  }

  @Override
  public FTPClient makeObject() throws Exception {
    FTPClient ftpClient = new FTPClient();
    ftpClient.setControlEncoding(ftpProperties.getEncoding());
    ftpClient.setConnectTimeout(ftpProperties.getClientTimeout());
    try {
      ftpClient.connect(ftpProperties.getHost(), ftpProperties.getPort());
      int reply = ftpClient.getReplyCode();
      if (!FTPReply.isPositiveCompletion(reply)) {
        ftpClient.disconnect();
        log.warn("FTPServer refused connection");
        return null;
      }
      boolean result = ftpClient.login(ftpProperties.getUsername(), ftpProperties.getPassword());
      ftpClient.setFileType(ftpProperties.getTransferFileType());
      if (!result) {
        log.warn("ftpClient login failed... username is {}", ftpProperties.getUsername());
      }
    } catch (Exception e) {
      log.error("create ftp connection failed...{}", e);
      throw e;
    }

    return ftpClient;
  }

  @Override
  public void destroyObject(FTPClient ftpClient) throws Exception {
    try {
      if(ftpClient != null && ftpClient.isConnected()) {
        ftpClient.logout();
      }
    } catch (Exception e) {
      log.error("ftp client logout failed...{}", e);
      throw e;
    } finally {
      if(ftpClient != null) {
        ftpClient.disconnect();
      }
    }

  }

  @Override
  public boolean validateObject(FTPClient ftpClient) {
    try {
      return ftpClient.sendNoOp();
    } catch (Exception e) {
      log.error("Failed to validate client: {}");
    }
    return false;
  }

  @Override
  public void activateObject(FTPClient obj) throws Exception {
    //Do nothing

  }

  @Override
  public void passivateObject(FTPClient obj) throws Exception {
    //Do nothing

  }
}

創(chuàng)建FTPUtils.java的工具類

FTPUtils.java中封裝了上傳、下載等方法，在項(xiàng)目啟動(dòng)的時(shí)候，在@PostConstruct注解的作用下通過(guò)執(zhí)行init()的方法，創(chuàng)建FTPClientFactory工廠中，并初始化了FTPClientPool線程池，這樣每次調(diào)用方法的時(shí)候，都直接從FTPClientPool中取出一個(gè)FTPClient對(duì)象

/**
 * @Auther: hrabbit
 * @Date: 2018-12-03 3:47 PM
 * @Description:
 */
@Slf4j
@Component
public class FTPUtils {

  /**
   * FTP的連接池
   */
  @Autowired
  public static FTPClientPool ftpClientPool;
  /**
   * FTPClient對(duì)象
   */
  public static FTPClient ftpClient;


  private static FTPUtils ftpUtils;

  @Autowired
  private FTPProperties ftpProperties;

  /**
   * 初始化設(shè)置
   * @return
   */
  @PostConstruct
  public boolean init() {
    FTPClientFactory factory = new FTPClientFactory(ftpProperties);
    ftpUtils = this;
    try {
      ftpClientPool = new FTPClientPool(factory);
    } catch (Exception e) {
      e.printStackTrace();
      return false;
    }
    return true;
  }


  /**
   * 獲取連接對(duì)象
   * @return
   * @throws Exception
   */
  public static FTPClient getFTPClient() throws Exception {
    //初始化的時(shí)候從隊(duì)列中取出一個(gè)連接
    if (ftpClient==null) {
      synchronized (ftpClientPool) {
        ftpClient = ftpClientPool.borrowObject();
      }
    }
    return ftpClient;
  }


  /**
   * 當(dāng)前命令執(zhí)行完成命令完成
   * @throws IOException
   */
  public void complete() throws IOException {
    ftpClient.completePendingCommand();
  }

  /**
   * 當(dāng)前線程任務(wù)處理完成，加入到隊(duì)列的最后
   * @return
   */
  public void disconnect() throws Exception {
    ftpClientPool.addObject(ftpClient);
  }

  /**
   * Description: 向FTP服務(wù)器上傳文件
   *
   * @Version1.0
   * @param remoteFile
   *      上傳到FTP服務(wù)器上的文件名
   * @param input
   *      本地文件流
   * @return 成功返回true，否則返回false
   */
  public static boolean uploadFile(String remoteFile, InputStream input) {
    boolean result = false;
    try {
      getFTPClient();
      ftpClient.enterLocalPassiveMode();
      result = ftpClient.storeFile(remoteFile, input);
      input.close();
      ftpClient.disconnect();
    } catch (Exception e) {
      e.printStackTrace();
    }
    return result;
  }

  /**
   * Description: 向FTP服務(wù)器上傳文件
   *
   * @Version1.0
   * @param remoteFile
   *      上傳到FTP服務(wù)器上的文件名
   * @param localFile
   *      本地文件
   * @return 成功返回true，否則返回false
   */
  public static boolean uploadFile(String remoteFile, String localFile){
    FileInputStream input = null;
    try {
      input = new FileInputStream(new File(localFile));
    } catch (FileNotFoundException e) {
      e.printStackTrace();
    }
    return uploadFile(remoteFile, input);
  }

  /**
   * 拷貝文件
   * @param fromFile
   * @param toFile
   * @return
   * @throws Exception
   */
  public boolean copyFile(String fromFile, String toFile) throws Exception {
    InputStream in=getFileInputStream(fromFile);
    getFTPClient();
    boolean flag = ftpClient.storeFile(toFile, in);
    in.close();
    return flag;
  }

  /**
   * 獲取文件輸入流
   * @param fileName
   * @return
   * @throws IOException
   */
  public static InputStream getFileInputStream(String fileName) throws Exception {
    ByteArrayOutputStream fos=new ByteArrayOutputStream();
    getFTPClient();
    ftpClient.retrieveFile(fileName, fos);
    ByteArrayInputStream in=new ByteArrayInputStream(fos.toByteArray());
    fos.close();
    return in;
  }

  /**
   * Description: 從FTP服務(wù)器下載文件
   *
   * @Version1.0
   * @return
   */
  public static boolean downFile(String remoteFile, String localFile){
    boolean result = false;
    try {
      getFTPClient();
      OutputStream os = new FileOutputStream(localFile);
      ftpClient.retrieveFile(remoteFile, os);
      ftpClient.logout();
      ftpClient.disconnect();
      result = true;
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      try {
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
    return result;
  }

  /**
   * 從ftp中獲取文件流
   * @param filePath
   * @return
   * @throws Exception
   */
  public static InputStream getInputStream(String filePath) throws Exception {
    getFTPClient();
    InputStream inputStream = ftpClient.retrieveFileStream(filePath);
    return inputStream;
  }

  /**
   * ftp中文件重命名
   * @param fromFile
   * @param toFile
   * @return
   * @throws Exception
   */
  public boolean rename(String fromFile,String toFile) throws Exception {
    getFTPClient();
    boolean result = ftpClient.rename(fromFile,toFile);
    return result;
  }

  /**
   * 獲取ftp目錄下的所有文件
   * @param dir
   * @return
   */
  public FTPFile[] getFiles(String dir) throws Exception {
    getFTPClient();
    FTPFile[] files = new FTPFile[0];
    try {
      files = ftpClient.listFiles(dir);
    }catch (Throwable thr){
      thr.printStackTrace();
    }
    return files;
  }

  /**
   * 獲取ftp目錄下的某種類型的文件
   * @param dir
   * @param filter
   * @return
   */
  public FTPFile[] getFiles(String dir, FTPFileFilter filter) throws Exception {
    getFTPClient();
    FTPFile[] files = new FTPFile[0];
    try {
      files = ftpClient.listFiles(dir, filter);
    }catch (Throwable thr){
      thr.printStackTrace();
    }
    return files;
  }

  /**
   * 創(chuàng)建文件夾
   * @param remoteDir
   * @return 如果已經(jīng)有這個(gè)文件夾返回false
   */
  public boolean makeDirectory(String remoteDir) throws Exception {
    getFTPClient();
    boolean result = false;
    try {
      result = ftpClient.makeDirectory(remoteDir);
    } catch (IOException e) {
      e.printStackTrace();
    }
    return result;
  }

  public boolean mkdirs(String dir) throws Exception {
    boolean result = false;
    if (null == dir) {
      return result;
    }
    getFTPClient();
    ftpClient.changeWorkingDirectory("/");
    StringTokenizer dirs = new StringTokenizer(dir, "/");
    String temp = null;
    while (dirs.hasMoreElements()) {
      temp = dirs.nextElement().toString();
      //創(chuàng)建目錄
      ftpClient.makeDirectory(temp);
      //進(jìn)入目錄
      ftpClient.changeWorkingDirectory(temp);
      result = true;
    }
    ftpClient.changeWorkingDirectory("/");
    return result;
  }
}

創(chuàng)建FtpClientTest.java測(cè)試類

上傳一張圖片到FTP服務(wù)器，并將文件重新命名為hrabbit.jpg，代碼如下:

/**
 * FtpClient測(cè)試
 * @Auther: hrabbit
 * @Date: 2018-12-21 9:14 PM
 * @Description:
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class FtpClientTest {

  /**
   * 測(cè)試上傳
   */
  @Test
  public void uploadFile(){
    boolean flag = FTPUtils.uploadFile("hrabbit.jpg", "/Users/mrotaku/Downloads/klklklkl_4x.jpg");
    Assert.assertEquals(true, flag);
  }

}

程序完美運(yùn)行，這時(shí)候我們查看我們的FTP服務(wù)器，http://localhost:8866/hrabbit.jpg