使用countDownLatch完成多线程任务的阻塞等待

标签：使用   post   执行   city   new t   dna   ack   tac   任务   

我这里需要通过多线程去处理数据，然后在所有数据都处理完成后再往下执行。这里就用到了CountDownLatch。把countdownlatch作为参数传入到每个线程类里，在线程中处理完数据后执行countdown方法。在所有countdownlatch归零后，其await方法结束阻塞状态而往下执行。

具体代码如下：

1. 将多线程任务提交线程池

         @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("=======上次抓取尚未结束=========");
   		}
   	}
   

2. 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();
   		}
   	}
   

使用countDownLatch完成多线程任务的阻塞等待

标签：使用   post   执行   city   new t   dna   ack   tac   任务   

原文地址：https://www.cnblogs.com/ayueC/p/13722747.html