溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
最近在寫一個FTP上傳工具，用到了Apache的FTPClient，為了提高上傳效率，我采用了多線程的方式，但是每個線程頻繁的創建和銷毀FTPClient對象勢必會造成不必要的開銷，因此，此處最好使用一個FTPClient連接池。仔細翻了一下Apache的api，發現它并沒有一個FTPClientPool的實現，所以，不得不自己寫一個FTPClientPool。下面就大體介紹一下開發連接池的整個過程，供大家參考。
關于對象池
有些對象的創建開銷是比較大的，比如數據庫連接等。為了減少頻繁創建、銷毀對象帶來的性能消耗，我們可以利用對象池的技術來實現對象的復用。對象池提供了一種機制，它可以管理對象池中對象的生命周期，提供了獲取和釋放對象的方法，可以讓客戶端很方便的使用對象池中的對象。
如果我們要自己實現一個對象池，一般需要完成如下功能：
1. 如果池中有可用的對象，對象池應當能返回給客戶端
2. 客戶端把對象放回池里后，可以對這些對象進行重用
3. 對象池能夠創建新的對象來滿足客戶端不斷增長的需求
4. 需要有一個正確關閉池的機制來結束對象的生命周期
Apache的對象池工具包
為了方便我們開發自己的對象池，Apache 提供的common-pool工具包，里面包含了開發通用對象池的一些接口和實現類。其中最基本的兩個接口是ObjectPool 和PoolableObjectFactory。
ObjectPool接口中有幾個最基本的方法：
1. addObject() : 添加對象到池
2. borrowObject()：客戶端從池中借出一個對象
3. returnObject()：客戶端歸還一個對象到池中
4. close()：關閉對象池，清理內存釋放資源等
5. setFactory(ObjectFactory factory)：需要一個工廠來制造池中的對象
PoolableObjectFactory接口中幾個最基本的方法：
1. makeObject()：制造一個對象
2. destoryObject()：銷毀一個對象
3. validateObject()：驗證一個對象是否還可用
通過以上兩個接口我們就可以自己實現一個對象池了。
實例：開發一個FTPClient對象池
最近在開發一個項目，需要把hdfs中的文件上傳到一組ftp服務器，為了提高上傳效率，自然考慮到使用多線程的方式進行上傳。我上傳ftp用的工具是Apache common-net包中的FTPClient，但Apache并沒有提供FTPClientPool，于是為了減少FTPClient的創建銷毀次數，我們就自己開發一個FTPClientPool來復用FTPClient連接。
通過上面的介紹，我們可以利用Apache提供的common-pool包來協助我們開發連接池。而開發一個簡單的對象池，僅需要實現common-pool 包中的ObjectPool和PoolableObjectFactory兩個接口即可。下面就看一下我寫的實現：
寫一個ObjectPool接口的實現FTPClientPool
import java.io.IOException;
import java.util.NoSuchElementException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.pool.ObjectPool;
import org.apache.commons.pool.PoolableObjectFactory;
/**
* 實現了一個FTPClient連接池
* @author heaven
*/
public class FTPClientPool implements ObjectPool<FTPClient>{
private static final int DEFAULT_POOL_SIZE = 10;
private final BlockingQueue<FTPClient> pool;
private final FtpClientFactory factory;
/**
* 初始化連接池，需要注入一個工廠來提供FTPClient實例
* @param factory
* @throws Exception
*/
public FTPClientPool(FtpClientFactory factory) throws Exception{
this(DEFAULT_POOL_SIZE, factory);
}
/**
*
* @param maxPoolSize
* @param factory
* @throws Exception
*/
public FTPClientPool(int poolSize, FtpClientFactory factory) throws Exception {
this.factory = factory;
pool = new ArrayBlockingQueue<FTPClient>(poolSize*2);
initPool(poolSize);
}
/**
* 初始化連接池，需要注入一個工廠來提供FTPClient實例
* @param maxPoolSize
* @throws Exception
*/
private void initPool(int maxPoolSize) throws Exception {
for(int i=0;i<maxPoolSize;i++){
//往池中添加對象
addObject();
}
}
/* (non-Javadoc)
* @see org.apache.commons.pool.ObjectPool#borrowObject()
*/
public FTPClient borrowObject() throws Exception, NoSuchElementException, IllegalStateException {
FTPClient client = pool.take();
if (client == null) {
client = factory.makeObject();
addObject();
}else if(!factory.validateObject(client)){//驗證不通過
//使對象在池中失效
invalidateObject(client);
//制造并添加新對象到池中
client = factory.makeObject();
addObject();
}
return client;
}
/* (non-Javadoc)
* @see org.apache.commons.pool.ObjectPool#returnObject(java.lang.Object)
*/
public void returnObject(FTPClient client) throws Exception {
if ((client != null) && !pool.offer(client,3,TimeUnit.SECONDS)) {
try {
factory.destroyObject(client);
} catch (IOException e) {
e.printStackTrace();
}
}
}
public void invalidateObject(FTPClient client) throws Exception {
//移除無效的客戶端
pool.remove(client);
}
/* (non-Javadoc)
* @see org.apache.commons.pool.ObjectPool#addObject()
*/
public void addObject() throws Exception, IllegalStateException, UnsupportedOperationException {
//插入對象到隊列
pool.offer(factory.makeObject(),3,TimeUnit.SECONDS);
}
public int getNumIdle() throws UnsupportedOperationException {
return 0;
}
public int getNumActive() throws UnsupportedOperationException {
return 0;
}
public void clear() throws Exception, UnsupportedOperationException {
}
/* (non-Javadoc)
* @see org.apache.commons.pool.ObjectPool#close()
*/
public void close() throws Exception {
while(pool.iterator().hasNext()){
FTPClient client = pool.take();
factory.destroyObject(client);
}
}
public void setFactory(PoolableObjectFactory<FTPClient> factory) throws IllegalStateException, UnsupportedOperationException {
}
}
再寫一個PoolableObjectFactory接口的實現FTPClientFactory
import java.io.IOException;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.commons.pool.PoolableObjectFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.hdfstoftp.util.FTPClientException;
/**
* FTPClient工廠類，通過FTPClient工廠提供FTPClient實例的創建和銷毀
* @author heaven
*/
public class FtpClientFactory implements PoolableObjectFactory<FTPClient> {
private static Logger logger = LoggerFactory.getLogger("file");
private FTPClientConfigure config;
//給工廠傳入一個參數對象，方便配置FTPClient的相關參數
public FtpClientFactory(FTPClientConfigure config){
this.config=config;
}
/* (non-Javadoc)
* @see org.apache.commons.pool.PoolableObjectFactory#makeObject()
*/
public FTPClient makeObject() throws Exception {
FTPClient ftpClient = new FTPClient();
ftpClient.setConnectTimeout(config.getClientTimeout());
try {
ftpClient.connect(config.getHost(), config.getPort());
int reply = ftpClient.getReplyCode();
if (!FTPReply.isPositiveCompletion(reply)) {
ftpClient.disconnect();
logger.warn("FTPServer refused connection");
return null;
}
boolean result = ftpClient.login(config.getUsername(), config.getPassword());
if (!result) {
throw new FTPClientException("ftpClient登陸失敗! userName:" + config.getUsername() + " ; password:" + config.getPassword());
}
ftpClient.setFileType(config.getTransferFileType());
ftpClient.setBufferSize(1024);
ftpClient.setControlEncoding(config.getEncoding());
if (config.getPassiveMode().equals("true")) {
ftpClient.enterLocalPassiveMode();
}
} catch (IOException e) {
e.printStackTrace();
} catch (FTPClientException e) {
e.printStackTrace();
}
return ftpClient;
}
/* (non-Javadoc)
* @see org.apache.commons.pool.PoolableObjectFactory#destroyObject(java.lang.Object)
*/
public void destroyObject(FTPClient ftpClient) throws Exception {
try {
if (ftpClient != null && ftpClient.isConnected()) {
ftpClient.logout();
}
} catch (IOException io) {
io.printStackTrace();
} finally {
// 注意,一定要在finally代碼中斷開連接，否則會導致占用ftp連接情況
try {
ftpClient.disconnect();
} catch (IOException io) {
io.printStackTrace();
}
}
}
/* (non-Javadoc)
* @see org.apache.commons.pool.PoolableObjectFactory#validateObject(java.lang.Object)
*/
public boolean validateObject(FTPClient ftpClient) {
try {
return ftpClient.sendNoOp();
} catch (IOException e) {
throw new RuntimeException("Failed to validate client: " + e, e);
}
}
public void activateObject(FTPClient ftpClient) throws Exception {
}
public void passivateObject(FTPClient ftpClient) throws Exception {
}
}
最后，我們最好給工廠傳遞一個參數對象，方便我們設置FTPClient的一些參數
package org.apache.commons.pool.impl.contrib;
/**
* FTPClient配置類，封裝了FTPClient的相關配置
*
* @author heaven
*/
public class FTPClientConfigure {
private String host;
private int port;
private String username;
private String password;
private String passiveMode;
private String encoding;
private int clientTimeout;
private int threadNum;
private int transferFileType;
private boolean renameUploaded;
private int retryTimes;
public String getHost() {
return host;
}
public void setHost(String host) {
this. host = host;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this. port = port;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this. username = username;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this. password = password;
}
public String getPassiveMode() {
return passiveMode;
}
public void setPassiveMode(String passiveMode) {
this. passiveMode = passiveMode;
}
public String getEncoding() {
return encoding;
}
public void setEncoding(String encoding) {
this. encoding = encoding;
}
public int getClientTimeout() {
return clientTimeout;
}
public void setClientTimeout( int clientTimeout) {
this. clientTimeout = clientTimeout;
}
public int getThreadNum() {
return threadNum;
}
public void setThreadNum( int threadNum) {
this. threadNum = threadNum;
}
public int getTransferFileType() {
return transferFileType;
}
public void setTransferFileType( int transferFileType) {
this. transferFileType = transferFileType;
}
public boolean isRenameUploaded() {
return renameUploaded;
}
public void setRenameUploaded( boolean renameUploaded) {
this. renameUploaded = renameUploaded;
}
public int getRetryTimes() {
return retryTimes;
}
public void setRetryTimes( int retryTimes) {
this. retryTimes = retryTimes;
}
@Override
public String toString() {
return "FTPClientConfig [host=" + host + "\n port=" + port + "\n username=" + username + "\n password=" + password + "\n passiveMode=" + passiveMode
+ "\n encoding=" + encoding + "\n clientTimeout=" + clientTimeout + "\n threadNum=" + threadNum + "\n transferFileType="
+ transferFileType + "\n renameUploaded=" + renameUploaded + "\n retryTimes=" + retryTimes + "]" ;
}
}
FTPClientPool連接池類管理FTPClient對象的生命周期，負責對象的借出、規劃、池的銷毀等；FTPClientPool類依賴于FtpClientFactory類，由這個工程類來制造和銷毀對象；FtpClientFactory又依賴FTPClientConfigure類，FTPClientConfigure負責封裝FTPClient的配置參數。至此，我們的FTPClient連接池就開發完成了。
需要注意的是，FTPClientPool中用到了一個阻塞隊列ArrayBlockingQueue來管理存放FTPClient對象，關于阻塞隊列，請參考我的這篇文章： 【Java并發之】BlockingQueue
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持億速云。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
