Java 如何快速實現一個連接池
ACP 庫提供了一整套用于實現對象池化的 API，以及若干種各具特色的對象池實現。目前最常用的版本是 2.0 版本，相對于 1.x 版本而言，并不是簡單升級。2.0 版本是對象池實現的完全重寫，顯著的提升了性能和可伸縮性，并且包含可靠的實例跟蹤和池監控。
Apache Commons Pool 的官網地址為：Pool ? Overview，想翻找相關文檔資料，到這里去是最權威、最全面的。
如何使用 ACP？要使用 ACP 實現一個線程池，首先需要先引入 ACP 的依賴包，這里以 Maven 為例。
<dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId><version>2.0</version></dependency>
要使用 ACP 實現一個對象池，大致可以分為三個步驟：
創建對象工廠：告訴 ACP 如何創建你要的對象。 創建對象池：告訴 ACP 你想創建一個怎樣的對象池。 使用對象池：ACP 告訴你如何使用你的對象。 創建對象工廠對象工廠告訴 ACP，它應該如何去創建、激活、鈍化、銷毀你的對象。創建對象工廠非常簡單，只需要實現 ACP 的 PooledObjectFactory 接口即可。PooledObjectFactory 接口的定義如下：
public interface PooledObjectFactory<T> { PooledObject<T> makeObject() throws Exception; void destroyObject(PooledObject<T> p) throws Exception; boolean validateObject(PooledObject<T> p); void activateObject(PooledObject<T> p) throws Exception; void passivateObject(PooledObject<T> p) throws Exception;}
但更多情況下，我們會繼承 BasePooledObjectFactory 類來實現對象工廠。因為 BasePooledObjectFactory 類是 PooledObjectFactory 的基礎實現類，使用它可以幫我們省了很多麻煩。通過繼承這個抽象類，我們只需要實現兩個方法：create() 和 wrap() 方法。
// 告訴 ACP 如何創建對象public abstract T create() throws Exception;// 定義你要返回的對象public abstract PooledObject<T> wrap(T obj);
create() 方法定義你的對象初始化過程，最后將初始化完成的對象返回。例如你想定義一個 SFTP 的連接，那么你首先需要定義一個 JSch 對象，之后設置賬號密碼，之后連接服務器，最后返回一個 ChannelSftp 對象。
public ChannelSftp create() { // SFTP 連接的創建過程}
wrap() 方法定義你要返回的對象，對于一個 SFTP 的連接池來說，其實就是一個 ChannelSftp 對象。一般情況下可以使用類 DefaultPooledObject 替代，參考實現如下：
@Overridepublic PooledObject<Foo> wrap(Foo foo) { return new DefaultPooledObject<Foo>(foo);}創建對象池
創建好對象工廠之后，ACP 已經知道你需要的對象如何創建了。那么接下來，你需要根據你的實際需要，去創建一個對象池。在 ACP 中，我們通過 GenericObjectPool 以及 GenericObjectPoolConfig 來創建一個對象池。
// 聲明一個對象池private GenericObjectPool<ChannelSftp> sftpConnectPool;// 設置連接池配置GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();poolConfig.setEvictionPolicyClassName('tech.shuyi.javacodechip.acp.SftpEvictionPolicy');poolConfig.setBlockWhenExhausted(true);poolConfig.setJmxEnabled(false);poolConfig.setMaxWaitMillis(1000 * 10);poolConfig.setTimeBetweenEvictionRunsMillis(60 * 1000);poolConfig.setMinEvictableIdleTimeMillis(20 * 1000);poolConfig.setTestWhileIdle(true);poolConfig.setTestOnReturn(true);poolConfig.setTestOnBorrow(true);poolConfig.setMaxTotal(3);// 設置拋棄策略AbandonedConfig abandonedConfig = new AbandonedConfig();abandonedConfig.setRemoveAbandonedOnMaintenance(true);abandonedConfig.setRemoveAbandonedOnBorrow(true);this.sftpConnectPool = new GenericObjectPool<>(sftpConnectFactory, poolConfig, abandonedConfig);
在上面創建 SFTP 連接池的代碼中，我們配置了一些線程池的參數以及設置了拋棄策略。拋棄策略是非常重要的，如果沒有設置拋棄策略，那么會拿到失效的連接從而導致獲取文件失敗。拋棄策略是通過 poolConfig.setEvictionPolicyClassName 來設置的，我們這里設置的是 SftpEvictionPolicy 類，其代碼內容如下：
@Slf4j@Componentpublic class SftpEvictionPolicy implements EvictionPolicy<com.jcraft.jsch.ChannelSftp> { @Override public boolean evict(EvictionConfig config, PooledObject<com.jcraft.jsch.ChannelSftp> underTest, int idleCount) {try { // 連接失效時進行驅逐 if (!underTest.getObject().isConnected()) {log.warn('connect time out, evict the connection. time={}',System.currentTimeMillis() - underTest.getLastReturnTime());return true; }}catch (Exception e){ return true;}return false; }}
看到這里，創建線程池的代碼就結束了，SftpConnectPool 文件的全部內容如下：
@Slf4jpublic class SftpConnectPool { private GenericObjectPool<ChannelSftp> sftpConnectPool; public SftpConnectPool(SftpConnectFactory sftpConnectFactory) {// 設置連接池配置GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();poolConfig.setEvictionPolicyClassName('tech.shuyi.javacodechip.acp.SftpEvictionPolicy');poolConfig.setBlockWhenExhausted(true);poolConfig.setJmxEnabled(false);poolConfig.setMaxWaitMillis(1000 * 10);poolConfig.setTimeBetweenEvictionRunsMillis(60 * 1000);poolConfig.setMinEvictableIdleTimeMillis(20 * 1000);poolConfig.setTestWhileIdle(true);poolConfig.setTestOnReturn(true);poolConfig.setTestOnBorrow(true);poolConfig.setMaxTotal(3);// 設置拋棄策略AbandonedConfig abandonedConfig = new AbandonedConfig();abandonedConfig.setRemoveAbandonedOnMaintenance(true);abandonedConfig.setRemoveAbandonedOnBorrow(true);this.sftpConnectPool = new GenericObjectPool<>(sftpConnectFactory, poolConfig, abandonedConfig); } public ChannelSftp borrowObject() {try { return sftpConnectPool.borrowObject();} catch (Exception e) { log.error('borrowObject error', e); return null;} } public void returnObject(ChannelSftp channelSftp) {if (channelSftp!=null) { sftpConnectPool.returnObject(channelSftp);} }}
為了方便使用，我還增加了 borrowObject 和 returnObject 方法，但這兩個并不是必須的。在這兩個方法中，我們分別調用了 GenericObjectPool 類的 borrowObject 方法和 returnObject 方法。這正是 ACP 提供的、使用線程池對象的方法，先借一個對象，之后歸還對象。
注：其實在這一步，已經包含了對象池的使用了。但實際使用的時候，我們經常是將對象池的聲明與使用放在同一個類中，因此為了講解方便，這里沒有分開。因此下文的使用對象池，本質上是對對象池做進一步封裝。
使用對象池到這里我們的 SFTP 對象池就已經創建完畢了，是不是非常簡單呢！但在實際的工作中，我們通常會在這基礎上，做一些封裝。對于我們這次的 SFTP 連接池來說，我們會對外直接提供下載文件的服務，將 SFTP 對象池進一步封裝起來，不需要關心怎么獲取文件。
public class SftpFileHelper { @Autowired private SftpConnectPool sftpConnectPool; public void download(String dir, String file, String saveUrl)throws IOException {ChannelSftp sftp = sftpConnectPool.borrowObject();log.info('begin to download file, dir={}, file={}, saveUrl={}', dir, file, saveUrl);try { if (!StringUtils.isEmpty(dir)) {sftp.cd(dir); } File downloadFile = new File(saveUrl); sftp.get(file, new FileOutputStream(downloadFile));}catch (Exception e){ log.warn('下載文件失敗', e);}finally { sftpConnectPool.returnObject(sftp);}log.info('file:{} is download successful', file); }}
最后我們寫一個測試用例來試一試，是否能正常下載文件。
@RunWith(SpringRunner.class)@SpringBootTest@Slf4jpublic class SftpFileHelperTest { @Autowired private SftpFileHelper sftpFileHelper; @Test public void testDownloadFtpFile() throws Exception {sftpFileHelper.download('dir', 'fileName', 'fileName'); }}總結
本文針對 Apache Commons Pool 庫最常用的對象池功能做了演示。看完這篇文章，我們知道創建一個線程池需要三個步驟，分別是：
創建對象工廠：告訴 ACP 如何創建你要的對象。 創建對象池：告訴 ACP 你想創建一個怎樣的對象池、設置驅逐策略。 使用對象池：ACP 告訴你如何使用你的對象。本文相關代碼存放在博主 Github 項目：java-code-chip 中，可以點擊地址獲取：java-code-chip/src/main/java/tech/shuyi/javacodechip/acp at master · chenyurong/java-code-chip
ACP 庫能夠讓讀者朋友們快速地創建一個對象池，更加專注于業務內容。但事實上，ACP 提供的內容遠不止如此，它還有更多更高級的功能。
例如當我們連接的 SFTP 服務器有多個時，我們需要通過不同地址來獲得不同的連接對象。此時最笨的辦法是每個不同的地址，都復制多一份代碼，然后通過不同類的不同方法來實現。但這樣的情況工作量相當可觀，并且也會有很多重復代碼。這種時候就可以使用BaseKeyedPooledObjectFactory 來替代 BasePooledObjectFactory，從而實現通過 key 來實現不同地址的連接對象管理。
更多關于 ACP 的內容，感興趣的同學可以自行探索，這里就不深入講解了。
另一種實現方式：泛型接口ConnectionPool.java
public interface ConnectionPool<T> { /** * 初始化池資源 * @param maxActive 池中最大活動連接數 * @param maxWait 最大等待時間 */ void init(Integer maxActive, Long maxWait); /** * 從池中獲取資源 * @return 連接資源 */ T getResource() throws Exception; /** * 釋放連接 * @param connection 正在使用的連接 */ void release(T connection) throws Exception; /** * 釋放連接池資源 */ void close();}
以zookeeper為例，實現zookeeper連接池，ZookeeperConnectionPool.java
public class ZookeeperConnectionPool implements ConnectionPool<ZooKeeper> { //最大活動連接數 private Integer maxActive; //最大等待時間 private Long maxWait; //空閑隊列 private LinkedBlockingQueue<ZooKeeper> idle = new LinkedBlockingQueue<>(); //繁忙隊列 private LinkedBlockingQueue<ZooKeeper> busy = new LinkedBlockingQueue<>(); //連接池活動連接數 private AtomicInteger activeSize = new AtomicInteger(0); //連接池關閉標記 private AtomicBoolean isClosed = new AtomicBoolean(false); //總共獲取的連接記數 private AtomicInteger createCount = new AtomicInteger(0); //等待zookeeper客戶端創建完成的計數器 private static ThreadLocal<CountDownLatch> latchThreadLocal = ThreadLocal.withInitial(() -> new CountDownLatch(1)); public ZookeeperConnectionPool(Integer maxActive, Long maxWait) {this.init(maxActive, maxWait); } @Override public void init(Integer maxActive, Long maxWait) {this.maxActive = maxActive;this.maxWait = maxWait; } @Override public ZooKeeper getResource() throws Exception {ZooKeeper zooKeeper;Long nowTime = System.currentTimeMillis();final CountDownLatch countDownLatch = latchThreadLocal.get();//空閑隊列idle是否有連接if ((zooKeeper = idle.poll()) == null) { //判斷池中連接數是否小于maxActive if (activeSize.get() < maxActive) {//先增加池中連接數后判斷是否小于等于maxActiveif (activeSize.incrementAndGet() <= maxActive) { //創建zookeeper連接 zooKeeper = new ZooKeeper('localhost', 5000, (watch) -> {if (watch.getState() == Watcher.Event.KeeperState.SyncConnected) { countDownLatch.countDown();} }); countDownLatch.await(); System.out.println('Thread:' + Thread.currentThread().getId() + '獲取連接：' + createCount.incrementAndGet() + '條'); busy.offer(zooKeeper); return zooKeeper;} else { //如增加后發現大于maxActive則減去增加的 activeSize.decrementAndGet();} } //若活動線程已滿則等待busy隊列釋放連接 try {System.out.println('Thread:' + Thread.currentThread().getId() + '等待獲取空閑資源');Long waitTime = maxWait - (System.currentTimeMillis() - nowTime);zooKeeper = idle.poll(waitTime, TimeUnit.MILLISECONDS); } catch (InterruptedException e) {throw new Exception('等待異常'); } //判斷是否超時 if (zooKeeper != null) {System.out.println('Thread:' + Thread.currentThread().getId() + '獲取連接：' + createCount.incrementAndGet() + '條');busy.offer(zooKeeper);return zooKeeper; } else {System.out.println('Thread:' + Thread.currentThread().getId() + '獲取連接超時，請重試！');throw new Exception('Thread:' + Thread.currentThread().getId() + '獲取連接超時，請重試！'); }}//空閑隊列有連接，直接返回busy.offer(zooKeeper);return zooKeeper; } @Override public void release(ZooKeeper connection) throws Exception {if (connection == null) { System.out.println('connection 為空'); return;}if (busy.remove(connection)){ idle.offer(connection);} else { activeSize.decrementAndGet(); throw new Exception('釋放失敗');} } @Override public void close() {if (isClosed.compareAndSet(false, true)) { idle.forEach((zooKeeper) -> {try { zooKeeper.close();} catch (InterruptedException e) { e.printStackTrace();} }); busy.forEach((zooKeeper) -> {try { zooKeeper.close();} catch (InterruptedException e) { e.printStackTrace();} });} }}測試用例
這里創建20個線程并發測試連接池，Test.java
public class Test { public static void main(String[] args) throws Exception {int threadCount = 20;Integer maxActive = 10;Long maxWait = 10000L;ZookeeperConnectionPool pool = new ZookeeperConnectionPool(maxActive, maxWait);CountDownLatch countDownLatch = new CountDownLatch(20);for (int i = 0; i < threadCount; i++) { new Thread(() -> {countDownLatch.countDown();try { countDownLatch.await(); ZooKeeper zooKeeper = pool.getResource(); Thread.sleep(2000); pool.release(zooKeeper);} catch (InterruptedException e) { e.printStackTrace();} catch (Exception e) { e.printStackTrace();} }).start();}while (true){} }}
以上就是Java 如何快速實現一個連接池的詳細內容，更多關于Java 連接池的資料請關注好吧啦網其它相關文章！
相關文章：
網公網安備