update 0303

03concurrency/0301/src/main/java/java0/conc0302/atomic/AtomicCount.java

@@ -7,7 +7,7 @@ public class AtomicCount {
 
     private AtomicInteger num = new AtomicInteger();
 
-    public synchronized int add() {
+    public int add() {
         return num.getAndIncrement();
     }
 

03concurrency/0301/src/main/java/java0/conc0302/lock/ReentrantReadWriteLockDemo2.java

@@ -41,7 +41,7 @@ public Object readWrite(String key) {
 
     public static void main(String[] args) {
         ReentrantReadWriteLockDemo2 demo2 = new ReentrantReadWriteLockDemo2();
-        demo2.readWrite("wangwei");
+        demo2.readWrite("bingfabiancheng");
     }
 
 }

03concurrency/0301/src/main/java/java0/conc0303/future/CompletableFutureDemo.java

@@ -0,0 +1,81 @@
+package java0.conc0303.future;
+
+import java.util.concurrent.CompletableFuture;
+
+public class CompletableFutureDemo {
+
+    public static void main(String[] args){
+
+        // 1.变换结果
+        System.out.println("=====>1.变换结果");
+        String result1 = CompletableFuture.supplyAsync(()->{return "Hello ";}).thenApplyAsync(v -> v + "world").join();
+        System.out.println(result1);
+
+        // 2.消费
+        CompletableFuture.supplyAsync(()->{return "Hello ";}).thenAccept(v -> { System.out.println("=====>2.消费");System.out.println("consumer: " + v);});
+
+        // 3.组合
+        System.out.println("=====>3.组合");
+        String result3 = CompletableFuture.supplyAsync(()->{
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            return "Hello";
+        }).thenCombine(CompletableFuture.supplyAsync(()->{
+            try {
+                Thread.sleep(2000);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            return "world";
+        }),(s1,s2)->{return s1 + " " + s2;}).join();
+        System.out.println("thenCombine:"+result3);
+
+        CompletableFuture.supplyAsync(() -> "Hello, java course.")
+                .thenApply(String::toUpperCase).thenCompose(s -> CompletableFuture.supplyAsync(s::toLowerCase)).thenAccept(v -> { System.out.println("thenCompose:"+v);});
+
+        // 4.竞争
+        System.out.println("=====>4.竞争");
+        String result4 = CompletableFuture.supplyAsync(()->{
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            return "Hi Boy";
+        }).applyToEither(CompletableFuture.supplyAsync(()->{
+            try {
+                Thread.sleep(300);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            return "Hi Girl";
+        }),(s)->{return s;}).join();
+        System.out.println(result4);
+
+        // 5.补偿异常
+        System.out.println("=====>5.补偿异常");
+        String result5 = CompletableFuture.supplyAsync(()->{
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            if(true) {
+                throw new RuntimeException("exception test!");
+            }
+
+            return "Hi Boy";
+        }).exceptionally(e->{
+            System.out.println(e.getMessage());
+            return "Hello world!";
+        }).join();
+        System.out.println(result5);
+
+
+
+    }
+
+}

03concurrency/0301/src/main/java/java0/conc0303/future/FutureDemo1.java

@@ -0,0 +1,23 @@
+package java0.conc0303.future;
+
+import java.util.Random;
+import java.util.concurrent.*;
+
+public class FutureDemo1 {
+    public static void main(String[] args) {
+        ExecutorService executor = Executors.newCachedThreadPool();
+        Future<Integer> result = executor.submit(new Callable<Integer>() {
+            public Integer call() throws Exception {
+                return new Random().nextInt();
+            }
+        });
+        executor.shutdown();
+        try {
+            System.out.println("result:" + result.get());
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        } catch (ExecutionException e) {
+            e.printStackTrace();
+        }
+    }
+}

03concurrency/0301/src/main/java/java0/conc0303/future/FutureTask1.java

@@ -0,0 +1,35 @@
+package java0.conc0303.future;
+
+import java.util.Random;
+import java.util.concurrent.*;
+
+public class FutureTask1 {
+    public static void main(String[] args) {
+        //第一种方式
+        FutureTask<Integer> task = new FutureTask<Integer>(new Callable<Integer>() {
+            @Override
+            public Integer call() throws Exception {
+                return new Random().nextInt();
+            }
+        });
+        new Thread(task).start();
+        //第二种方方式
+//        ExecutorService executor = Executors.newSingleThreadExecutor();
+//        FutureTask<Integer> task = new FutureTask<Integer>(new Callable<Integer>() {
+//            @Override
+//            public Integer call() throws Exception {
+//                return new Random().nextInt();
+//            }
+//        });
+//        executor.submit(task);
+
+        try {
+            System.out.println("result: " + task.get());
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        } catch (ExecutionException e) {
+            e.printStackTrace();
+        }
+    }
+
+}

03concurrency/0301/src/main/java/java0/conc0303/tool/CountDownLatchDemo.java

@@ -0,0 +1,33 @@
+package java0.conc0303.tool;
+
+import java.util.concurrent.CountDownLatch;
+
+public class CountDownLatchDemo {
+    public static void main(String[] args) throws InterruptedException {
+        CountDownLatch countDownLatch = new CountDownLatch(5);
+        for(int i=0;i<5;i++){
+            new Thread(new readNum(i,countDownLatch)).start();
+        }
+        countDownLatch.await(); // 注意跟CyclicBarrier不同，这里在主线程await
+        System.out.println("==>各个子线程执行结束。。。。");
+        System.out.println("==>主线程执行结束。。。。");
+    }
+
+    static class readNum  implements Runnable{
+        private int id;
+        private CountDownLatch latch;
+        public readNum(int id,CountDownLatch latch){
+            this.id = id;
+            this.latch = latch;
+        }
+        @Override
+        public void run() {
+            synchronized (this){
+                System.out.println("id:"+id+","+Thread.currentThread().getName());
+                //latch.countDown();
+                System.out.println("线程组任务"+id+"结束，其他任务继续");
+                latch.countDown();
+            }
+        }
+    }
+}

03concurrency/0301/src/main/java/java0/conc0303/tool/CountDownLatchDemo2.java

@@ -0,0 +1,40 @@
+package java0.conc0303.tool;
+
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class CountDownLatchDemo2 {
+
+    private final static int threadCount = 200;
+
+    public static void main(String[] args) throws Exception {
+
+        ExecutorService exec = Executors.newCachedThreadPool();
+
+        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
+
+        for (int i = 0; i < threadCount; i++) {
+            final int threadNum = i;
+            exec.execute(() -> {
+                try {
+                    test(threadNum);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                } finally {
+                    countDownLatch.countDown();
+                }
+            });
+        }
+        countDownLatch.await();
+        System.out.println("==>所有程序员完成任务，项目顺利上线！");
+        exec.shutdown();
+    }
+
+    private static void test(int threadNum) throws Exception {
+        Thread.sleep(100);
+        System.out.println(String.format("程序员[%d]完成任务。。。", threadNum));
+        Thread.sleep(100);
+    }
+}

03concurrency/0301/src/main/java/java0/conc0303/tool/CyclicBarrierDemo.java

@@ -0,0 +1,48 @@
+package java0.conc0303.tool;
+
+import java.util.concurrent.CyclicBarrier;
+
+public class CyclicBarrierDemo {
+    public static void main(String[] args) throws InterruptedException {
+        CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() {
+            @Override
+            public void run() {
+                System.out.println("回调>>"+Thread.currentThread().getName());
+                System.out.println("回调>>线程组执行结束");
+            }
+        });
+        for (int i = 0; i < 5; i++) {
+            new Thread(new readNum(i,cyclicBarrier)).start();
+        }
+
+        System.out.println("==>各个子线程执行结束。。。。");
+        System.out.println("==>主线程执行结束。。。。");
+
+        //CyclicBarrier 可以重复利用，
+        // 这个是CountDownLatch做不到的
+//        for (int i = 11; i < 16; i++) {
+//            new Thread(new readNum(i,cyclicBarrier)).start();
+//        }
+    }
+    static class readNum  implements Runnable{
+        private int id;
+        private CyclicBarrier cyc;
+        public readNum(int id,CyclicBarrier cyc){
+            this.id = id;
+            this.cyc = cyc;
+        }
+        @Override
+        public void run() {
+            synchronized (this){
+                System.out.println("id:"+id+","+Thread.currentThread().getName());
+                try {
+                    //cyc.await();
+                    System.out.println("线程组任务" + id + "结束，其他任务继续");
+                    cyc.await();   // 注意跟CountDownLatch不同，这里在子线程await
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+}

03concurrency/0301/src/main/java/java0/conc0303/tool/CyclicBarrierDemo2.java

@@ -0,0 +1,49 @@
+package java0.conc0303.tool;
+
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+
+public class CyclicBarrierDemo2 {
+    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(10000);
+        } 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(3000);      //以睡眠来模拟写入数据操作
+                System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");
+
+                cyclicBarrier.await();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }catch(BrokenBarrierException e){
+                e.printStackTrace();
+            }
+            System.out.println(Thread.currentThread().getName()+"所有线程写入完毕，继续处理其他任务...");
+        }
+    }
+}

03concurrency/0301/src/main/java/java0/conc0303/tool/SemaphoreDemo.java

@@ -0,0 +1,36 @@
+package java0.conc0303.tool;
+
+import 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();
+    }
+
+    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();  // 在子线程里控制资源占用
+                System.out.println("工人" + this.num + "占用一个机器在生产...");
+                Thread.sleep(2000);
+                System.out.println("工人" + this.num + "释放出机器");
+                semaphore.release();   // 在子线程里控制释放资源占用
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+}