java多线程异步CountDownLatch

import java.io.BufferedWriter;import java.io.File;import java.io.FileWriter;import java.io.IOException;import java.util.Scanner;import java.util.concurrent.BlockingQueue;import java.util.concurrent.CountDownLatch;import java.util.concurrent.LinkedBlockingQueue;public class AsyncExecute {// (一)，用CountDownLatch控制多线程的执行顺序// (二)，多个线程同时工作时// (三)，但有一个线程需要等其他线程结束后才能开始 public AsyncExecute(CountDownLatch cdl) { this.latch = cdl; queue = new LinkedBlockingQueue(); File file = new File(“F:xbl1.sql”); try{ bw = new BufferedWriter(new FileWriter(file)); } catch(Exception ex) { ex.printStackTrace(); } } // 控制资源释放 private CountDownLatch latch; // 阻塞队列 private BlockingQueue queue; private BufferedWriter bw; private boolean finshedHandle = false; // 线程能发现volatile变量最新值 private volatile boolean sendFinished = false; public void sendFinish() { sendFinished = true; } // 释放资源 public void release() { System.out.println(“release..”); if (bw != null) { try { bw.close(); } catch (IOException e) { } } if(queue != null) { queue.clear(); queue = null; } } // 往队列发送消息 public void sendMsg(String text) { queue.add(text); } public void handle() { // 模拟性能瓶颈每3秒 new Thread( new Runnable() { public void run() { while(!finshedHandle) { try { Thread.sleep(3000); } catch(Exception ex) {} // 队列中获取 String s = queue !=null? queue.peek():null; if(s != null) { // remove弹出 queue.poll(); try { bw.write(s); bw.newLine(); } catch(Exception ex) {} } // 队列为空 if(sendFinished) { /* CountDownLatch 是一个同步工具类, 用来协作多个线程间的同步,或起到线程间通信 (而不是互斥)。 (一)，他是通过计数器来实现的， (二)，计数器的初始值为线程的数量 * (三)，每当一个线程完成任务后， (四)，计数器的值就会减1。 * (五)，当计数器的值为0时， (六)，它表示所有线程已经完成任务 */ // 计数器1->0 latch.countDown(); finshedHandle = true; break; } } } } ).start(); } public static void main(String[] args)throws Exception { //(一)，CountDownLatch这个类 //(二)，可以使一个线程等待 //(三)，其他线程完成任务后再执行 CountDownLatch latch = new CountDownLatch(1); AsyncExecute async = new AsyncExecute(latch); async.handle(); Scanner scanner = new Scanner(System.in); while(true) { String text = scanner.next(); if(text.equals(“exit”)) { async.sendFinish(); break; } async.sendMsg(text); } try { // 阻塞主线程,等待消息写入到本地文件完成 latch.wait(); } catch(Exception ex) {} async.release(); scanner.close(); } }