GitHub : https://github.com/jayknoxqu/ftp-pool
一. 連接池概述
頻繁的建立和關閉連接，會極大的降低系統的性能，而連接池會在初始化的時候會創建一定數量的連接，每次訪問只需從連接池里獲取連接,使用完畢后再放回連接池，並不是直接關閉連接，這樣可以保證程序重復使用同一個連接而不需要每次訪問都建立和關閉連接， 從而提高系統性能。
二. 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 (對象池)
對應為:
FTPClient(池化對象) FTPClientFactory(對象工廠) FTPClientPool(對象池)
關系圖:
三. 實現連接池
3.1 配置FtpClient
我們已經有現成的池化對象(FtpClient)了,只需要添加配置即可
@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配置為:
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這個抽象類 ,而它則實現了PooledObjectFactory
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); } } }
不太贊成自己去實現連接池,這樣會帶來額外的維護成本...
四. 代碼地址:
GitHub : https://github.com/jayknoxqu/ftp-pool
五. 參考資料:
commons-pool2 官方案列