Java countDownLatch如何實現多執行緒任務阻塞等待
阿新 • • 發佈:2020-10-15
我這裡需要通過多執行緒去處理資料,然後在所有資料都處理完成後再往下執行。這裡就用到了CountDownLatch。把countdownlatch作為引數傳入到每個執行緒類裡,線上程中處理完資料後執行countdown方法。在所有countdownlatch歸零後,其await方法結束阻塞狀態而往下執行。
具體程式碼如下:
將多執行緒任務提交執行緒池
@Bean(name = "ggnews_executor") public Executor postExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(1); executor.setMaxPoolSize(1); executor.setQueueCapacity(1); executor.setKeepAliveSeconds(120); executor.setThreadNamePrefix("executor-"); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy()); return executor; } //通過定時任務呼叫的fetch方法,為了避免定時任務在多次執行中失效,通非同步指定執行緒池的方式進行呼叫 @Async("ggnews_executor") public void fetch() { if(fetchFlag.getAndSet(false)) { List<FetchTag> tags = fetchTagService.selectFetchTagList(fetchTag); CountDownLatch downLatch = new CountDownLatch(tags.size()); for (FetchTag tag : tags) { FetchTag tagNew; try { tagNew =(FetchTag) tag.clone(); } catch (Throwable e) { log.error("",e); continue; } //作為引數將CountDownLatch傳入 InnerRunner innerRunner = new InnerRunner(downLatch,tagNew); executor.execute(innerRunner); } try { //等待執行緒執行完畢,如果十分鐘後還沒結束也會停止阻塞狀態 downLatch.await(10,TimeUnit.MINUTES); fetchFlag.getAndSet(true); } catch (Throwable e) { log.error("fetch()方法發生錯誤:{}",e); fetchFlag.getAndSet(true); //e.printStackTrace(); } finally { fetchFlag.getAndSet(true); } } else { log.info("=======上次抓取尚未結束========="); } }
InnerRunner為要執行具體任務的執行緒類
private class InnerRunner implements Runnable { private CountDownLatch downLatch; private FetchTag tag; private InnerRunner(CountDownLatch downLatch,FetchTag tag) { this.downLatch = downLatch; this.tag = tag; } @Override public void run() { //將countDown方法移入到具體方法中的finally塊中,以保證即使在丟擲異常的情況下也算執行了此次任務,countdown會被執行 fetchGG(tag.getTag(),downLatch); //downLatch.countDown(); this.tag = null; } }
private static final String GOOGLE_URL_IN = "https://news.google.com/rss/search?hl=hi&gl=IN&ceid=IN:hi&q="; public void fetchGG(String tag,CountDownLatch downLatch) { try { Document document = Jsoup.parse(new URL(GOOGLE_URL_IN + URLEncoder.encode("\"" + tag + "\"","utf-8")),30000); Elements elements = document.getElementsByTag("item"); int rank = 1; for (Element element : elements) { String sourceTitle = element.getElementsByTag("title").get(0).text(); log.info("source title:" + sourceTitle); } } catch (Throwable e) { log.info("fetch google url error",e); } finally { //肯定會被執行 downLatch.countDown(); } }
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支援我們。