使用commons-pool2實現FTP鏈接池

時間 2019-11-08

標籤使用 commons pool2 pool 實現 ftp 鏈接欄目 FTP 简体版

原文原文鏈接

GitHub : github.com/jayknoxqu/f…html

一. 鏈接池概述

頻繁的創建和關閉鏈接，會極大的下降系統的性能，而鏈接池會在初始化的時候會建立必定數量的鏈接，每次訪問只需從鏈接池裏獲取鏈接,使用完畢後再放回鏈接池，並非直接關閉鏈接，這樣能夠保證程序重複使用同一個鏈接而不須要每次訪問都創建和關閉鏈接，從而提升系統性能。java

二. commons-pool2介紹

2.1 pool2的引入

<!-- 使用commons-pool2 實現ftp鏈接池 -->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
    <version>2.5.0</version>
</dependency>

<!-- 引入FTPClient做爲池化對象 -->
<dependency>
    <groupId>commons-net</groupId>
    <artifactId>commons-net</artifactId>
    <version>3.6</version>
</dependency>
複製代碼

2.2 pool2的組成

PooledObject(池化對象) PooledObjectFactory(對象工廠) ObjectPool (對象池)git

對應爲:github

FTPClient(池化對象) FTPClientFactory(對象工廠) FTPClientPool(對象池)apache

關係圖:bash

三. 實現鏈接池

3.1 配置FtpClient

咱們已經有現成的池化對象(FtpClient)了,只須要添加配置便可app

@ConfigurationProperties(ignoreUnknownFields = false, prefix = "ftp.client")
public class FtpClientProperties {
    // ftp地址
    private String host;
    // 端口號
    private Integer port = 21;
    // 登陸用戶
    private String username;
    // 登陸密碼
    private String password;
    // 被動模式
    private boolean passiveMode = false;
    // 編碼
    private String encoding = "UTF-8";
    // 鏈接超時時間(秒)
    private Integer connectTimeout;
    // 緩衝大小
    private Integer bufferSize = 1024;
    // 傳輸文件類型
    private Integer transferFileType;
}
複製代碼

application.properties配置爲:ide

ftp.client.host=127.0.0.1
ftp.client.port=22
ftp.client.username=root
ftp.client.password=root
ftp.client.encoding=utf-8
ftp.client.passiveMode=false
ftp.client.connectTimeout=30000
複製代碼

3.2 建立FtpClientFactory

在commons-pool2中有兩種工廠：PooledObjectFactory 和KeyedPooledObjectFactory，咱們使用前者。性能

public interface PooledObjectFactory<T> {
	//建立對象
    PooledObject<T> makeObject();
	//激活對象
    void activateObject(PooledObject<T> obj);
    //鈍化對象
    void passivateObject(PooledObject<T> obj);
    //驗證對象
    boolean validateObject(PooledObject<T> obj);
    //銷燬對象
    void destroyObject(PooledObject<T> obj);
}
複製代碼

建立FtpClientFactory只須要繼承BasePooledObjectFactory這個抽象類 ,而它則實現了PooledObjectFactorythis

public class FtpClientFactory extends BasePooledObjectFactory<FTPClient> {

    private FtpClientProperties config;

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

    /** * 建立FtpClient對象 */
    @Override
    public FTPClient create() {
        FTPClient ftpClient = new FTPClient();
        ftpClient.setControlEncoding(config.getEncoding());
        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 (IOException e) {
            log.error("create ftp connection failed...", e);
        }
        return ftpClient;
    }

    /** * 用PooledObject封裝對象放入池中 */
    @Override
    public PooledObject<FTPClient> wrap(FTPClient ftpClient) {
        return new DefaultPooledObject<>(ftpClient);
    }

    /** * 銷燬FtpClient對象 */
    @Override
    public void destroyObject(PooledObject<FTPClient> ftpPooled) {
        if (ftpPooled == null) {
            return;
        }

        FTPClient ftpClient = ftpPooled.getObject();

        try {
            if (ftpClient.isConnected()) {
                ftpClient.logout();
            }
        } catch (IOException io) {
            log.error("ftp client logout failed...{}", io);
        } finally {
            try {
                ftpClient.disconnect();
            } catch (IOException io) {
                log.error("close ftp client failed...{}", io);
            }
        }
    }

    /** * 驗證FtpClient對象 */
    @Override
    public boolean validateObject(PooledObject<FTPClient> ftpPooled) {
        try {
            FTPClient ftpClient = ftpPooled.getObject();
            return ftpClient.sendNoOp();
        } catch (IOException e) {
            log.error("Failed to validate client: {}", e);
        }
        return false;
    }

}
複製代碼

3.3 實現FtpClientPool

在commons-pool2中預設了三個能夠直接使用的對象池：GenericObjectPool、GenericKeyedObjectPool和SoftReferenceObjectPool

示列:

GenericObjectPool<FTPClient> ftpClientPool = new GenericObjectPool<>(new FtpClientFactory());
複製代碼

咱們也能夠本身實現一個鏈接池:

public interface ObjectPool<T> extends Closeable {
    // 從池中獲取一個對象
    T borrowObject();
	// 歸還一個對象到池中
    void returnObject(T obj);
    // 廢棄一個失效的對象
    void invalidateObject(T obj); 
    // 添加對象到池
    void addObject();
	// 清空對象池
    void clear();
    // 關閉對象池
    void close();
}
複製代碼

經過繼承BaseObjectPool去實現ObjectPool

public class FtpClientPool extends BaseObjectPool<FTPClient> {

    private static final int DEFAULT_POOL_SIZE = 8;

    private final BlockingQueue<FTPClient> ftpBlockingQueue;
    private final FtpClientFactory ftpClientFactory;


    /** * 初始化鏈接池，須要注入一個工廠來提供FTPClient實例 * * @param ftpClientFactory ftp工廠 * @throws Exception */
    public FtpClientPool(FtpClientFactory ftpClientFactory) throws Exception {
        this(DEFAULT_POOL_SIZE, ftpClientFactory);
    }

    public FtpClientPool(int poolSize, FtpClientFactory factory) throws Exception {
        this.ftpClientFactory = factory;
        ftpBlockingQueue = new ArrayBlockingQueue<>(poolSize);
        initPool(poolSize);
    }

    /** * 初始化鏈接池，須要注入一個工廠來提供FTPClient實例 * * @param maxPoolSize 最大鏈接數 * @throws Exception */
    private void initPool(int maxPoolSize) throws Exception {
        for (int i = 0; i < maxPoolSize; i++) {
            // 往池中添加對象
            addObject();
        }
    }

    /** * 從鏈接池中獲取對象 */
    @Override
    public FTPClient borrowObject() throws Exception {
        FTPClient client = ftpBlockingQueue.take();
        if (ObjectUtils.isEmpty(client)) {
            client = ftpClientFactory.create();
            // 放入鏈接池
            returnObject(client);
            // 驗證對象是否有效
        } else if (!ftpClientFactory.validateObject(ftpClientFactory.wrap(client))) {
            // 對無效的對象進行處理
            invalidateObject(client);
            // 建立新的對象
            client = ftpClientFactory.create();
            // 將新的對象放入鏈接池
            returnObject(client);
        }
        return client;
    }

    /** * 返還對象到鏈接池中 */
    @Override
    public void returnObject(FTPClient client) {
        try {
            if (client != null && !ftpBlockingQueue.offer(client, 3, TimeUnit.SECONDS)) {
                ftpClientFactory.destroyObject(ftpClientFactory.wrap(client));
            }
        } catch (InterruptedException e) {
            log.error("return ftp client interrupted ...{}", e);
        }
    }

    /** * 移除無效的對象 */
    @Override
    public void invalidateObject(FTPClient client) {
        try {
            client.changeWorkingDirectory("/");
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            ftpBlockingQueue.remove(client);
        }
    }

    /** * 增長一個新的連接，超時失效 */
    @Override
    public void addObject() throws Exception {
        // 插入對象到隊列
        ftpBlockingQueue.offer(ftpClientFactory.create(), 3, TimeUnit.SECONDS);
    }

    /** * 關閉鏈接池 */
    @Override
    public void close() {
        try {
            while (ftpBlockingQueue.iterator().hasNext()) {
                FTPClient client = ftpBlockingQueue.take();
                ftpClientFactory.destroyObject(ftpClientFactory.wrap(client));
            }
        } catch (Exception e) {
            log.error("close ftp client ftpBlockingQueue failed...{}", e);
        }
    }

}
複製代碼