av一区二区在线观看_亚洲男人的天堂网站_日韩亚洲视频_在线成人免费_欧美日韩精品免费观看视频_久草视

您的位置:首頁技術(shù)文章
文章詳情頁

Java 如何快速實(shí)現(xiàn)一個(gè)連接池

瀏覽:4日期:2022-08-11 16:45:46
目錄什么是 ACP?如何使用 ACP?創(chuàng)建對象工廠創(chuàng)建對象池使用對象池總結(jié)另一種實(shí)現(xiàn)方式:測試用例什么是 ACP?

ACP 庫提供了一整套用于實(shí)現(xiàn)對象池化的 API,以及若干種各具特色的對象池實(shí)現(xiàn)。目前最常用的版本是 2.0 版本,相對于 1.x 版本而言,并不是簡單升級。2.0 版本是對象池實(shí)現(xiàn)的完全重寫,顯著的提升了性能和可伸縮性,并且包含可靠的實(shí)例跟蹤和池監(jiān)控。

Apache Commons Pool 的官網(wǎng)地址為:Pool ? Overview,想翻找相關(guān)文檔資料,到這里去是最權(quán)威、最全面的。

如何使用 ACP?

要使用 ACP 實(shí)現(xiàn)一個(gè)線程池,首先需要先引入 ACP 的依賴包,這里以 Maven 為例。

<dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId><version>2.0</version></dependency>

要使用 ACP 實(shí)現(xiàn)一個(gè)對象池,大致可以分為三個(gè)步驟:

創(chuàng)建對象工廠:告訴 ACP 如何創(chuàng)建你要的對象。 創(chuàng)建對象池:告訴 ACP 你想創(chuàng)建一個(gè)怎樣的對象池。 使用對象池:ACP 告訴你如何使用你的對象。 創(chuàng)建對象工廠

對象工廠告訴 ACP,它應(yīng)該如何去創(chuàng)建、激活、鈍化、銷毀你的對象。創(chuàng)建對象工廠非常簡單,只需要實(shí)現(xiàn) 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;}

但更多情況下,我們會(huì)繼承 BasePooledObjectFactory 類來實(shí)現(xiàn)對象工廠。因?yàn)?BasePooledObjectFactory 類是 PooledObjectFactory 的基礎(chǔ)實(shí)現(xiàn)類,使用它可以幫我們省了很多麻煩。通過繼承這個(gè)抽象類,我們只需要實(shí)現(xiàn)兩個(gè)方法:create() 和 wrap() 方法。

// 告訴 ACP 如何創(chuàng)建對象public abstract T create() throws Exception;// 定義你要返回的對象public abstract PooledObject<T> wrap(T obj);

create() 方法定義你的對象初始化過程,最后將初始化完成的對象返回。例如你想定義一個(gè) SFTP 的連接,那么你首先需要定義一個(gè) JSch 對象,之后設(shè)置賬號密碼,之后連接服務(wù)器,最后返回一個(gè) ChannelSftp 對象。

public ChannelSftp create() { // SFTP 連接的創(chuàng)建過程}

wrap() 方法定義你要返回的對象,對于一個(gè) SFTP 的連接池來說,其實(shí)就是一個(gè) ChannelSftp 對象。一般情況下可以使用類 DefaultPooledObject 替代,參考實(shí)現(xiàn)如下:

@Overridepublic PooledObject<Foo> wrap(Foo foo) { return new DefaultPooledObject<Foo>(foo);}創(chuàng)建對象池

創(chuàng)建好對象工廠之后,ACP 已經(jīng)知道你需要的對象如何創(chuàng)建了。那么接下來,你需要根據(jù)你的實(shí)際需要,去創(chuàng)建一個(gè)對象池。在 ACP 中,我們通過 GenericObjectPool 以及 GenericObjectPoolConfig 來創(chuàng)建一個(gè)對象池。

// 聲明一個(gè)對象池private GenericObjectPool<ChannelSftp> sftpConnectPool;// 設(shè)置連接池配置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);// 設(shè)置拋棄策略AbandonedConfig abandonedConfig = new AbandonedConfig();abandonedConfig.setRemoveAbandonedOnMaintenance(true);abandonedConfig.setRemoveAbandonedOnBorrow(true);this.sftpConnectPool = new GenericObjectPool<>(sftpConnectFactory, poolConfig, abandonedConfig);

在上面創(chuàng)建 SFTP 連接池的代碼中,我們配置了一些線程池的參數(shù)以及設(shè)置了拋棄策略。拋棄策略是非常重要的,如果沒有設(shè)置拋棄策略,那么會(huì)拿到失效的連接從而導(dǎo)致獲取文件失敗。拋棄策略是通過 poolConfig.setEvictionPolicyClassName 來設(shè)置的,我們這里設(shè)置的是 SftpEvictionPolicy 類,其代碼內(nèi)容如下:

@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 { // 連接失效時(shí)進(jìn)行驅(qū)逐 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; }}

看到這里,創(chuàng)建線程池的代碼就結(jié)束了,SftpConnectPool 文件的全部內(nèi)容如下:

@Slf4jpublic class SftpConnectPool { private GenericObjectPool<ChannelSftp> sftpConnectPool; public SftpConnectPool(SftpConnectFactory sftpConnectFactory) {// 設(shè)置連接池配置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);// 設(shè)置拋棄策略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 方法,但這兩個(gè)并不是必須的。在這兩個(gè)方法中,我們分別調(diào)用了 GenericObjectPool 類的 borrowObject 方法和 returnObject 方法。這正是 ACP 提供的、使用線程池對象的方法,先借一個(gè)對象,之后歸還對象。

注:其實(shí)在這一步,已經(jīng)包含了對象池的使用了。但實(shí)際使用的時(shí)候,我們經(jīng)常是將對象池的聲明與使用放在同一個(gè)類中,因此為了講解方便,這里沒有分開。因此下文的使用對象池,本質(zhì)上是對對象池做進(jìn)一步封裝。

使用對象池

到這里我們的 SFTP 對象池就已經(jīng)創(chuàng)建完畢了,是不是非常簡單呢!但在實(shí)際的工作中,我們通常會(huì)在這基礎(chǔ)上,做一些封裝。對于我們這次的 SFTP 連接池來說,我們會(huì)對外直接提供下載文件的服務(wù),將 SFTP 對象池進(jìn)一步封裝起來,不需要關(guān)心怎么獲取文件。

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); }}

最后我們寫一個(gè)測試用例來試一試,是否能正常下載文件。

@RunWith(SpringRunner.class)@SpringBootTest@Slf4jpublic class SftpFileHelperTest { @Autowired private SftpFileHelper sftpFileHelper; @Test public void testDownloadFtpFile() throws Exception {sftpFileHelper.download('dir', 'fileName', 'fileName'); }}總結(jié)

本文針對 Apache Commons Pool 庫最常用的對象池功能做了演示。看完這篇文章,我們知道創(chuàng)建一個(gè)線程池需要三個(gè)步驟,分別是:

創(chuàng)建對象工廠:告訴 ACP 如何創(chuàng)建你要的對象。 創(chuàng)建對象池:告訴 ACP 你想創(chuàng)建一個(gè)怎樣的對象池、設(shè)置驅(qū)逐策略。 使用對象池:ACP 告訴你如何使用你的對象。

本文相關(guān)代碼存放在博主 Github 項(xiàng)目:java-code-chip 中,可以點(diǎn)擊地址獲取:java-code-chip/src/main/java/tech/shuyi/javacodechip/acp at master · chenyurong/java-code-chip

ACP 庫能夠讓讀者朋友們快速地創(chuàng)建一個(gè)對象池,更加專注于業(yè)務(wù)內(nèi)容。但事實(shí)上,ACP 提供的內(nèi)容遠(yuǎn)不止如此,它還有更多更高級的功能。

例如當(dāng)我們連接的 SFTP 服務(wù)器有多個(gè)時(shí),我們需要通過不同地址來獲得不同的連接對象。此時(shí)最笨的辦法是每個(gè)不同的地址,都復(fù)制多一份代碼,然后通過不同類的不同方法來實(shí)現(xiàn)。但這樣的情況工作量相當(dāng)可觀,并且也會(huì)有很多重復(fù)代碼。這種時(shí)候就可以使用BaseKeyedPooledObjectFactory 來替代 BasePooledObjectFactory,從而實(shí)現(xiàn)通過 key 來實(shí)現(xiàn)不同地址的連接對象管理。

更多關(guān)于 ACP 的內(nèi)容,感興趣的同學(xué)可以自行探索,這里就不深入講解了。

另一種實(shí)現(xiàn)方式:

泛型接口ConnectionPool.java

public interface ConnectionPool&lt;T&gt; { /** * 初始化池資源 * @param maxActive 池中最大活動(dòng)連接數(shù) * @param maxWait 最大等待時(shí)間 */ void init(Integer maxActive, Long maxWait); /** * 從池中獲取資源 * @return 連接資源 */ T getResource() throws Exception; /** * 釋放連接 * @param connection 正在使用的連接 */ void release(T connection) throws Exception; /** * 釋放連接池資源 */ void close();}

以zookeeper為例,實(shí)現(xiàn)zookeeper連接池,ZookeeperConnectionPool.java

public class ZookeeperConnectionPool implements ConnectionPool&lt;ZooKeeper&gt; { //最大活動(dòng)連接數(shù) private Integer maxActive; //最大等待時(shí)間 private Long maxWait; //空閑隊(duì)列 private LinkedBlockingQueue&lt;ZooKeeper&gt; idle = new LinkedBlockingQueue&lt;&gt;(); //繁忙隊(duì)列 private LinkedBlockingQueue&lt;ZooKeeper&gt; busy = new LinkedBlockingQueue&lt;&gt;(); //連接池活動(dòng)連接數(shù) private AtomicInteger activeSize = new AtomicInteger(0); //連接池關(guān)閉標(biāo)記 private AtomicBoolean isClosed = new AtomicBoolean(false); //總共獲取的連接記數(shù) private AtomicInteger createCount = new AtomicInteger(0); //等待zookeeper客戶端創(chuàng)建完成的計(jì)數(shù)器 private static ThreadLocal&lt;CountDownLatch&gt; latchThreadLocal = ThreadLocal.withInitial(() -&gt; 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();//空閑隊(duì)列idle是否有連接if ((zooKeeper = idle.poll()) == null) { //判斷池中連接數(shù)是否小于maxActive if (activeSize.get() &lt; maxActive) {//先增加池中連接數(shù)后判斷是否小于等于maxActiveif (activeSize.incrementAndGet() &lt;= maxActive) { //創(chuàng)建zookeeper連接 zooKeeper = new ZooKeeper('localhost', 5000, (watch) -&gt; {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 { //如增加后發(fā)現(xiàn)大于maxActive則減去增加的 activeSize.decrementAndGet();} } //若活動(dòng)線程已滿則等待busy隊(duì)列釋放連接 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('等待異常'); } //判斷是否超時(shí) 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() + '獲取連接超時(shí),請重試!');throw new Exception('Thread:' + Thread.currentThread().getId() + '獲取連接超時(shí),請重試!'); }}//空閑隊(duì)列有連接,直接返回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) -&gt; {try { zooKeeper.close();} catch (InterruptedException e) { e.printStackTrace();} }); busy.forEach((zooKeeper) -&gt; {try { zooKeeper.close();} catch (InterruptedException e) { e.printStackTrace();} });} }}測試用例

這里創(chuàng)建20個(gè)線程并發(fā)測試連接池,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 &lt; threadCount; i++) { new Thread(() -&gt; {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 如何快速實(shí)現(xiàn)一個(gè)連接池的詳細(xì)內(nèi)容,更多關(guān)于Java 連接池的資料請關(guān)注好吧啦網(wǎng)其它相關(guān)文章!

標(biāo)簽: Java
相關(guān)文章:
主站蜘蛛池模板: 尹人av| 欧美人妇做爰xxxⅹ性高电影 | 久久久青草 | 国产亚洲一区在线 | 91在线精品视频 | 欧美欧美欧美 | 福利成人 | 久久国产精品偷 | 久久夜色精品国产 | av中文字幕网 | www.99热| 久久综合一区 | 欧美男人天堂 | 久久这里有精品 | 91大神在线资源观看无广告 | 亚洲一区二区在线视频 | 久久小视频| 亚洲一区二区三区免费在线观看 | 日美女逼逼 | 日韩国产欧美一区 | 久久久久久久久国产成人免费 | 国产精品久久精品 | 国产高清一二三区 | 亚洲一区在线日韩在线深爱 | 日本a级大片 | 中文字幕第十页 | 99在线国产 | 精品欧美一区二区在线观看 | 日韩三级免费网站 | 久久精品一 | 国产精品中文字幕在线播放 | 日韩一区二区三区精品 | 国产精品久久久久久久久久久免费看 | 国产成人午夜精品影院游乐网 | 夜夜久久 | 玖玖综合在线 | 激情小视频 | 亚洲看片 | 欧美三级在线 | 亚洲成人一区二区三区 | 天天操天天干天天爽 |