最近在寫一個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
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
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
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
|
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的一些參數
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
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
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
原文鏈接：https://blog.csdn.net/suifeng3051/article/details/48969679
