🔖 并发工具类示例

dunwu · dunwu · commit 6466abc80c08 · 2018-05-15T19:16:54.000+08:00
diff --git a/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/CountDownLatchDemo.java b/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/CountDownLatchDemo.java
@@ -4,8 +4,11 @@
 
 /**
  * CountDownLatch 示例
- * CountDownLatch 可以实现类似计数器的功能。
- *
+ * <p>
+ * 作用：允许一个或多个线程等待，直到在其他线程中执行的一组操作完成。
+ * <p>
+ * 原理：CountDownLatch 维护一个计数器 count。每次调用 countDown 方法会让 count 的值减 1，减到 0 的时候，那些因为调用 await
+ * 方法而在等待的线程就会被唤醒。
  * @author Zhang Peng
  * @date 2018/5/10
  * @see java.util.concurrent.CountDownLatch
diff --git a/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/CountDownLatchDemo02.java b/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/CountDownLatchDemo02.java
@@ -0,0 +1,54 @@
+package io.github.dunwu.javase.concurrent.tool;
+
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * @author Zhang Peng
+ * @date 2018/5/15
+ */
+public class CountDownLatchDemo02 {
+    public static void main(String[] args) throws InterruptedException {
+        Runnable task = new ThreadName();
+        long time = timeTasks(3, task);
+        System.out.println(time);
+    }
+
+    private static long timeTasks(int num, final Runnable task)
+        throws InterruptedException {
+        final CountDownLatch startGate = new CountDownLatch(1);
+        final CountDownLatch endGate = new CountDownLatch(num);
+
+        for (int i = 0; i < num; i++) {
+            Thread t = new Thread(() -> {
+                try {
+                    startGate.await();
+                    try {
+                        task.run();
+                    } finally {
+                        endGate.countDown();
+                    }
+                } catch (InterruptedException ignored) {
+                }
+            });
+            t.start();
+        }
+
+        long start = System.currentTimeMillis();
+        startGate.countDown();
+        endGate.await();
+        long end = System.currentTimeMillis();
+        return end - start;
+    }
+
+    static class ThreadName implements Runnable {
+
+        @Override
+        public void run() {
+            for (int i = 1; i <= 5; i++) {
+                System.out.println(Thread.currentThread().getName() + "运行，i = " + i);
+            }
+        }
+    }
+
+    ;
+}
diff --git a/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/CyclicBarrierDemo.java b/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/CyclicBarrierDemo.java
@@ -7,7 +7,6 @@
  * CyclicBarrier 示例
  * 字面意思回环栅栏，通过它可以实现让一组线程等待至某个状态之后再全部同时执行。
  * 叫做回环是因为当所有等待线程都被释放以后，CyclicBarrier可以被重用。
- *
  * @author Zhang Peng
  * @date 2018/5/10
  * @see java.util.concurrent.CyclicBarrier
diff --git a/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/CyclicBarrierDemo02.java b/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/CyclicBarrierDemo02.java
@@ -9,52 +9,52 @@
  * CyclicBarrier 示例
  * 字面意思回环栅栏，通过它可以实现让一组线程等待至某个状态之后再全部同时执行。
  * 叫做回环是因为当所有等待线程都被释放以后，CyclicBarrier可以被重用。
- *
  * @author Zhang Peng
  * @date 2018/5/10
  * @see CyclicBarrier
  */
 public class CyclicBarrierDemo02 {
 
-    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 Writer(barrier).start(); } else {
-                try {
-                    Thread.sleep(5000);
-                } catch (InterruptedException e) {
-                    e.printStackTrace();
-                }
-                new Writer(barrier).start();
-            }
-        }
-    }
+    static class CyclicBarrierRunnable implements Runnable {
 
-    static class Writer extends Thread {
+        CyclicBarrier barrier1 = null;
+        CyclicBarrier barrier2 = null;
 
-        private CyclicBarrier cyclicBarrier;
-
-        Writer(CyclicBarrier cyclicBarrier) {
-            this.cyclicBarrier = cyclicBarrier;
+        CyclicBarrierRunnable(CyclicBarrier barrier1, CyclicBarrier barrier2) {
+            this.barrier1 = barrier1;
+            this.barrier2 = barrier2;
         }
 
-        @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) {
-                    e.printStackTrace();
-                }
+                Thread.sleep(1000);
+                System.out.println(Thread.currentThread().getName() + " waiting at barrier 1");
+                this.barrier1.await();
+
+                Thread.sleep(1000);
+                System.out.println(Thread.currentThread().getName() + " waiting at barrier 2");
+                this.barrier2.await();
+
+                System.out.println(Thread.currentThread().getName() + " done!");
+
             } catch (InterruptedException | BrokenBarrierException e) {
                 e.printStackTrace();
             }
-            System.out.println(Thread.currentThread().getName() + "所有线程写入完毕，继续处理其他任务...");
         }
     }
+
+    public static void main(String[] args) {
+        Runnable barrier1Action = () -> System.out.println("BarrierAction 1 executed ");
+        Runnable barrier2Action = () -> System.out.println("BarrierAction 2 executed ");
+
+        CyclicBarrier barrier1 = new CyclicBarrier(2, barrier1Action);
+        CyclicBarrier barrier2 = new CyclicBarrier(2, barrier2Action);
+
+        CyclicBarrierRunnable barrierRunnable1 = new CyclicBarrierRunnable(barrier1, barrier2);
+
+        CyclicBarrierRunnable barrierRunnable2 = new CyclicBarrierRunnable(barrier1, barrier2);
+
+        new Thread(barrierRunnable1).start();
+        new Thread(barrierRunnable2).start();
+    }
 }
diff --git a/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/SemaphoreDemo.java b/codes/concurrent/src/main/java/io/github/dunwu/javase/concurrent/tool/SemaphoreDemo.java
@@ -1,45 +1,38 @@
 package io.github.dunwu.javase.concurrent.tool;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Semaphore;
 
 /**
  * Semaphore 示例
  * 字面意思为信号量，Semaphore 可以控同时访问的线程个数，通过 acquire() 获取一个许可，
  * 如果没有就等待，而 release() 释放一个许可。
- *
  * @author Zhang Peng
  * @date 2018/5/10
  * @see java.util.concurrent.Semaphore
  */
 public class SemaphoreDemo {
 
-    public static void main(String[] args) {
-        int N = 8;            //工人数
-        Semaphore semaphore = new Semaphore(5); //机器数目
-        for (int i = 0; i < N; i++) { new Worker(i, semaphore).start(); }
-    }
+    private static final int THREAD_COUNT = 30;
 
-    static class Worker extends Thread {
+    private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
 
-        private int num;
-        private Semaphore semaphore;
+    private static Semaphore s = new Semaphore(10);
 
-        Worker(int num, Semaphore semaphore) {
-            this.num = num;
-            this.semaphore = semaphore;
+    public static void main(String[] args) {
+        for (int i = 0; i < THREAD_COUNT; i++) {
+            threadPool.execute(() -> {
+                try {
+                    s.acquire();
+                    System.out.println("save data");
+                    s.release();
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                }
+            });
         }
 
-        @Override
-        public void run() {
-            try {
-                semaphore.acquire(); // 获取许可
-                System.out.println("工人" + this.num + "占用一个机器在生产...");
-                Thread.sleep(2000);
-                System.out.println("工人" + this.num + "释放出机器");
-                semaphore.release(); // 释放许可
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-        }
+        threadPool.shutdown();
     }
 }
