【同步器
java.util.concurrent包包含几个能帮助人们管理相互合作的线程集的类。这些机制具有为线程直间的共用集结点模式提供的‘预制功能’。如果有一个相互合作的线程满足这些行为模式之一,那么应该直接使用提供的类库而不是显示的使用锁与条件的集合。
【倒计时门栓
一个倒计时门栓(CountDownlatch)让一个线程集直到计数变为0.倒计时门栓是一次性的,一旦计数为0就不能再重用了。一个有用的特例是计数值为1的门栓。实现一个只能通过一次的门。线程在门外等待直到另一个线程将计数值变为0。举例来讲,假设一个线程集需要一些初始数据来完成工作。工作线程被启动并在,门外等候,另一个线程准备数据,当数据准备好时,调用countDown(),所有的工作线程就可以继续工作了。然后再使用一个门栓检查什么时候工作线程全部运行完成。每个工作线程在结束前将门栓计数器减一,门栓的计数变为0就表明工作完成。
【常用方法
- public void countDown():递减锁存器的计数,如果计数到达零,则释放所有等待的线程。如果当前计数大于零,则将计数减少。如果新的计数为零,出于线程调度目的,将重新启用所有的等待线程。如果当前计数等于零,则不发生任何操作。
- public boolean await():使当前线程在锁存器倒计数至零之前一直等待,如果当前计数为零,则此方法立刻返回 true 值。
-
public boolean await(long timeout,TimeUnit unit) throws InterruptedException:使当前线程在锁存器倒计数至零之前一直等待,除非线程被中断或超出了指定的等待时间。如果当前计数为零,则此方法立刻返回 true 值。如果当前计数大于零,则出于线程调度目的,将禁用当前线程,且在发生以下三种情况之一前,该线程将一直处于休眠状态:
- 由于调用 countDown() 方法,计数到达零;或者其他某个线程中断当前线程;或者已超出指定的等待时间。如果计数到达零,则该方法返回 true 值。
- 如果当前线程:在进入此方法时已经设置了该线程的中断状态;或者在等待时被中断,则抛出 InterruptedException,并且清除当前线程的已中断状态。
- 如果超出了指定的等待时间,则返回值为 false。如果该时间小于等于零,则此方法根本不会等待。
【例子
模拟一个应用程序:在正式开始工作前需要初始化数据,初始化数据使用三个线程,正式执行需要五个线程:
- 初始化线程
public class InitThread implements Runnable{ private CountDownLatch downLatch; private String name; public InitThread(CountDownLatch downLatch, String name){ this.downLatch = downLatch; this.name = name; } public void run() { this.doWork(); try{ TimeUnit.SECONDS.sleep(new Random().nextInt(10)); }catch(InterruptedException ie){ } System.out.println(this.name + "初始化数据完成"); //计数器减一 this.downLatch.countDown(); } private void doWork(){ System.out.println(this.name + "正在初始化数据... ..."); }}
- 初始化线程监视器
/** * 检测初始化数据监视器,因为需要判断是否初始化线程全部执行完毕,这里用callable返回结果。runnable不能返回值所以无法判断。 */public class InitMonitor implements Callable{ private ExecutorService executor; private CountDownLatch initLatch; private List initThreads; public InitMonitor(ExecutorService executor){ this.executor = executor; //初始化线程:3个 initLatch = new CountDownLatch(3); initThreads = Arrays.asList(new InitThread(initLatch,"InitOne"), new InitThread(initLatch,"InitTwo"), new InitThread(initLatch,"InitThree")); } public String call() { System.out.println("=========初始化START=========="); initThreads.stream().forEach(initThread -> executor.submit(initThread)); try { initLatch.await(); } catch (InterruptedException e) { } System.out.println("***********初始化END*************"); return "INIT_SUCCESS"; }}
- 工作线程
public class ExecuteThread implements Runnable{ private CountDownLatch downLatch; private String name; public ExecuteThread(CountDownLatch downLatch, String name){ this.downLatch = downLatch; this.name = name; } public void run() { this.doWork(); try{ TimeUnit.SECONDS.sleep(new Random().nextInt(10)); }catch(InterruptedException ie){ } System.out.println(this.name + "执行完成"); //计数器减一 this.downLatch.countDown(); } private void doWork(){ System.out.println(this.name + "正在执行... ..."); }}
- 工作线程监视器
public class ExecuteMonitor implements Callable{ private ExecutorService executor; private CountDownLatch executeLatch; private List executeThreads; public ExecuteMonitor(ExecutorService executor){ this.executor = executor; //执行线程:5个 executeLatch = new CountDownLatch(5); executeThreads = Arrays.asList(new ExecuteThread(executeLatch,"ExecuteOne"), new ExecuteThread(executeLatch,"ExecuteTwo"), new ExecuteThread(executeLatch,"ExecuteThree"), new ExecuteThread(executeLatch,"ExecuteFour"), new ExecuteThread(executeLatch,"ExecuteFive")); } public String call() { System.out.println("========执行START========"); executeThreads.stream().forEach(executeThread -> executor.submit(executeThread)); try { executeLatch.await(); } catch (InterruptedException e) { } System.out.println("*********执行END*********"); return "EXECUTE_SUCCESS"; }}
- 应用程序
public class Application implements Runnable{ private ExecutorService executor; private InitMonitor initMonitor; private ExecuteMonitor executeMonitor; public Application(ExecutorService executor){ this.executor = executor; initMonitor = new InitMonitor(executor); executeMonitor = new ExecuteMonitor(executor); } @Override public void run() { System.out.println("===============应用程序执行开始====================》》》"); FutureTaskinitTask = new FutureTask (initMonitor); executor.submit(initTask); try { //如果初始化成功开始执行工作线程,在调用get()时,如果没有执行完成会自动阻塞,所以这里不需要使用isDone检测。 if("INIT_SUCCESS".equals(initTask.get())){ FutureTask executeTask = new FutureTask (executeMonitor); executor.submit(executeTask); if("EXECUTE_SUCCESS".equals(executeTask.get())){ executor.shutdown(); System.out.println("===============应用程序执行完毕===================="); } } } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } }}
- 客户端
客户端一定尽量简介,所有细节全部屏蔽,这里只留下一个可以自定义线程池给用户自行选择
public class Test { public static void main(String[] args) throws InterruptedException { ExecutorService executor = Executors.newFixedThreadPool(10); Application application = new Application(executor); application.run(); }}