add CountDownLatch & CyclicBarrier & Semaphore

Author: maskleo

README.md

@@ -1,5 +1,34 @@
 # java-concurrent-programming-explain
 
+## Thread.join()
+
+- 子线程执行完成才执行主线程
+- [ThreadJoinTest](module1/src/edu/maskleo/module1/ThreadJoinTest.java)
+
+
+## CountDownLatch
+
+- 所有子线程完成了阻塞 `count` 计数之后就开始调用主线程的 `latch.await()` 之后的方法
+- [CountDownLatchTest](module1/src/edu/maskleo/module1/CyclicBarrierTest.java)
+
+## CyclicBarrier
+
+- 子线程完成等待之后，调用指定的线程执行,[示例](module1/src/edu/maskleo/module1/CyclicBarrierTest.java),一个类完成上述功能,[相关示例](module1/src/edu/maskleo/module1/CyclicBarrierTest2.java).
+
+## CountDownLatch & CyclicBarrier
+
+- `CountDownLatch` 和 `CyclicBarrier` 都能够实现线程之间的等待，只不过它们侧重点不同：
+- `CountDownLatch` 一般用于某个线程A等待若干个其他线程执行完任务之后，它才执行；
+  
+- 而 `CyclicBarrier` 一般用于一组线程互相等待至某个状态，然后这一组线程再同时执行；
+  
+- 另外，`CountDownLatch` 是不能够重用的，而 `CyclicBarrier` 是可以重用的。
+
+## Semaphore
+
+- Semaphore 其实和锁有点类似，它一般用于控制对某组资源的访问权限。
+- 初始化时设置大小, 线程每次获取和存放回去都有数量可选。
+
 ## 資料
 
 - [The j.u.c Synchronizer Framework中文翻译版](http://ifeve.com/aqs/)
@@ -10,6 +39,7 @@
 
 - [AQS解析(2)](https://ryan-hou.github.io/2018/06/13/AQS%E8%A7%A3%E6%9E%90-2/)
 
+- [CountDownLatch、CyclicBarrier和 Semaphore](http://www.importnew.com/21889.html)
 
 ## LICENSE
 

module1/src/edu/maskleo/module1/CountDownLatchTest.java

@@ -0,0 +1,46 @@
+package edu.maskleo.module1;
+
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+
+public class CountDownLatchTest {
+
+    public static void main(String[] args) {
+        CountDownLatch latch = new CountDownLatch(4);
+        Thread t1 = new Thread(new InnerThread(latch),"t1");
+        Thread t2 = new Thread(new InnerThread(latch),"t2");
+        Thread t3 = new Thread(new InnerThread(latch),"t3");
+        Thread t4 = new Thread(new InnerThread(latch),"t4");
+        t1.start();
+        t2.start();
+        t3.start();
+        t4.start();
+        try {
+            latch.await();
+        }catch (Exception e){
+            ;
+        }
+        System.out.println("all finish!");
+    }
+
+    static class InnerThread implements Runnable{
+
+        private CountDownLatch latch;
+
+        public InnerThread(CountDownLatch latch){
+            this.latch = latch;
+        }
+
+        @Override
+        public void run() {
+            System.out.println(Thread.currentThread().getName() + "begin to do something !");
+            try {
+                Thread.sleep(new Random().nextInt(1000));
+            }catch (Exception e){
+                ;
+            }
+            System.out.println(Thread.currentThread().getName() + "finish !");
+            latch.countDown();
+        }
+    }
+}

module1/src/edu/maskleo/module1/CyclicBarrierTest.java

@@ -0,0 +1,40 @@
+package edu.maskleo.module1;
+
+import java.util.concurrent.CyclicBarrier;
+
+public class CyclicBarrierTest {
+
+    public static void main(String[] args) {
+        CyclicBarrier barrier = new CyclicBarrier(4, () -> System.out.println("xxxxx"));
+        Thread t1 = new Thread(new InnerThread(barrier), "t1");
+        Thread t2 = new Thread(new InnerThread(barrier), "t2");
+        Thread t3 = new Thread(new InnerThread(barrier), "t3");
+        Thread t4 = new Thread(new InnerThread(barrier), "t4");
+        t1.start();
+        t2.start();
+        t3.start();
+        t4.start();
+        System.out.println("all finish!");
+    }
+
+    static class InnerThread implements Runnable{
+
+        private CyclicBarrier barrier;
+
+        public InnerThread(CyclicBarrier barrier){
+            this.barrier = barrier;
+        }
+
+        @Override
+        public void run() {
+            System.out.println(Thread.currentThread().getName() + " begin to do something !");
+            try {
+                barrier.await();
+            }catch (Exception e){
+                ;
+            }
+            System.out.println(Thread.currentThread().getName() + " finish !");
+        }
+    }
+
+}

module1/src/edu/maskleo/module1/CyclicBarrierTest2.java

@@ -0,0 +1,54 @@
+package edu.maskleo.module1;
+
+import java.util.Map;
+import java.util.concurrent.*;
+
+public class CyclicBarrierTest2 implements Runnable {
+    /**
+     * 创建4个屏障，处理完之后执行当前类的run方法
+     */
+    private CyclicBarrier c = new CyclicBarrier(4, this);
+    /**
+     * 假设只有4个sheet，所以只启动4个线程
+     */
+    private Executor executor = Executors.newFixedThreadPool(4);
+    /**
+     * 保存每个sheet计算出的银流结果
+     */
+    private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new
+            ConcurrentHashMap<>();
+
+    public static void main(String[] args) {
+        CyclicBarrierTest2 bankWaterCount = new CyclicBarrierTest2();
+        bankWaterCount.count();
+    }
+
+    private void count() {
+        for (int i = 0; i < 4; i++) {
+            executor.execute(() -> {
+                // 计算当前sheet的银流数据，计算代码省略
+                sheetBankWaterCount
+                        .put(Thread.currentThread().getName(), 1);
+                // 银流计算完成，插入一个屏障
+                try {
+                    c.await();
+                } catch (InterruptedException |
+                        BrokenBarrierException e) {
+                    e.printStackTrace();
+                }
+            });
+        }
+    }
+
+    @Override
+    public void run() {
+        int result = 0;
+        // 汇总每个sheet计算出的结果
+        for (Map.Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) {
+            result += sheet.getValue();
+        }
+        // 将结果输出
+        sheetBankWaterCount.put("result", result);
+        System.out.println(result);
+    }
+}

module1/src/edu/maskleo/module1/CyclicBarrierTest3.java

@@ -0,0 +1,55 @@
+package edu.maskleo.module1;
+
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class CyclicBarrierTest3 {
+
+    public static void main(String[] args) {
+        int N = 4;
+        CyclicBarrier barrier = new CyclicBarrier(N);
+
+        for (int i = 0; i < N; i++) {
+            if (i < N - 1)
+                new Thread(new Writer(barrier), "t" + i).start();
+            else {
+                try {
+                    Thread.sleep(5000);
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                }
+                new Thread(new Writer(barrier), "t" + i).start();
+            }
+        }
+    }
+
+    static class Writer implements Runnable {
+        private CyclicBarrier cyclicBarrier;
+
+        public Writer(CyclicBarrier cyclicBarrier) {
+            this.cyclicBarrier = cyclicBarrier;
+        }
+
+        @Override
+        public void run() {
+            System.out.println("线程" + Thread.currentThread().getName() + " 正在写入数据...");
+            try {
+                Thread.sleep(5000);      //以睡眠来模拟写入数据操作
+                System.out.println("线程" + Thread.currentThread().getName() + " 写入数据完毕，等待其他线程写入完毕");
+                try {
+                    cyclicBarrier.await(2000, TimeUnit.MILLISECONDS);
+                } catch (TimeoutException e) {
+                    System.out.println(Thread.currentThread().getName() + " 1111");
+                }
+            } catch (InterruptedException e) {
+                System.out.println(Thread.currentThread().getName() + " 2222");
+            } catch (BrokenBarrierException e) {
+                System.out.println(Thread.currentThread().getName() + " 3333");
+            }
+            System.out.println(Thread.currentThread().getName() + " 所有线程写入完毕，继续处理其他任务...");
+        }
+    }
+
+}

module1/src/edu/maskleo/module1/CyclicBarrierTest4.java

@@ -0,0 +1,53 @@
+package edu.maskleo.module1;
+
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+
+public class CyclicBarrierTest4 {
+
+    public static void main(String[] args) {
+        int N = 4;
+        CyclicBarrier barrier = new CyclicBarrier(N);
+
+        for (int i = 0; i < N; i++) {
+            new Writer(barrier).start();
+        }
+
+        try {
+            Thread.sleep(2500);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+
+        System.out.println("CyclicBarrier重用");
+
+        for (int i = 0; i < N; i++) {
+            new Writer(barrier).start();
+        }
+    }
+
+    static class Writer extends Thread {
+        private CyclicBarrier cyclicBarrier;
+
+        public Writer(CyclicBarrier cyclicBarrier) {
+            this.cyclicBarrier = cyclicBarrier;
+        }
+
+        @Override
+        public void run() {
+            System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
+            try {
+                Thread.sleep(500);      //以睡眠来模拟写入数据操作
+                System.out.println("线程" + Thread.currentThread().getName() + "写入数据完毕，等待其他线程写入完毕");
+
+                cyclicBarrier.await();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            } catch (BrokenBarrierException e) {
+                e.printStackTrace();
+            }
+            System.out.println(Thread.currentThread().getName() + "所有线程写入完毕，继续处理其他任务...");
+        }
+    }
+
+}

module1/src/edu/maskleo/module1/SemaphoreTest.java

@@ -0,0 +1,45 @@
+package edu.maskleo.module1;
+
+import java.util.Random;
+import java.util.concurrent.Semaphore;
+
+public class SemaphoreTest {
+
+    final static Object lock = new Object();
+
+    final static int NUM = 10;
+
+    public static void main(String[] args) {
+        int N = 10;            //工人数
+        Semaphore semaphore = new Semaphore(NUM); //机器数目
+        for (int i = 0; i < N; i++)
+            new Worker(i, semaphore).start();
+    }
+
+    static class Worker extends Thread {
+        private int num;
+        private Semaphore semaphore;
+
+        public Worker(int num, Semaphore semaphore) {
+            this.num = num;
+            this.semaphore = semaphore;
+        }
+
+        @Override
+        public void run() {
+            try {
+                semaphore.acquire(5);
+                System.out.println("工人" + this.num + "占用2个机器在生产...");
+                Thread.sleep(new Random().nextInt(10));
+                System.out.println("工人" + this.num + "释放出机器");
+                semaphore.release(2);
+                Thread.sleep(new Random().nextInt(1000));
+                System.out.println("工人" + this.num + "释放出机器");
+                semaphore.release(3);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+}

module1/src/edu/maskleo/module1/ThreadJoinTest.java

@@ -0,0 +1,44 @@
+package edu.maskleo.module1;
+
+import java.util.Random;
+
+public class ThreadJoinTest {
+
+    public static void main(String[] args) {
+        Thread t1 = new Thread(new InnerThread(),"t1");
+        Thread t2 = new Thread(new InnerThread(),"t2");
+        Thread t3 = new Thread(new InnerThread(),"t3");
+        Thread t4 = new Thread(new InnerThread(),"t4");
+        t1.start();
+        t2.start();
+        t3.start();
+        t4.start();
+        join(t1, t2, t3, t4);
+        System.out.println("all finish!");
+    }
+
+    private static void join(Thread t1, Thread t2, Thread t3, Thread t4) {
+        try {
+            t1.join();
+            t2.join();
+            t3.join();
+            t4.join();
+        }catch (Exception e){
+        }
+    }
+
+    static class InnerThread implements Runnable{
+        @Override
+        public void run() {
+            System.out.println(Thread.currentThread().getName() + "begin to do something !");
+            try {
+                Thread.sleep(new Random().nextInt(1000));
+            }catch (Exception e){
+                ;
+            }
+            System.out.println(Thread.currentThread().getName() + "finish !");
+        }
+    }
+
+
+}