JUC强大的辅助工具类

傷城~ 2022-08-28 13:50 185阅读 0赞

JUC中提供了一些辅助类，通过这些辅助类可以很好的解决线程数量过多时Lock锁的频繁操作。常用的三种辅助类有：

• CountDownLatch: 减少计数

• CyclicBarrier: 循环栅栏

• Semaphore: 信号灯

# **CountDownLatch** #

CountDownLatch是一个同步工具类，用来进行线程同步协作，等待所有线程完成倒计时。

CountDownLatch类可以设置一个计数器，然后通过countDown方法来进行减1的操作，使用await方法等待计数器不大于0，然后继续执行await方法之后的语句。

• CountDownLatch主要有两个方法，当一个或多个线程调用await方法时，这些线程会阻塞

• 其它线程调用countDown方法会将计数器减1(调用countDown方法的线程不会阻塞)

• 当计数器的值变为0时，因await方法阻塞的线程会被唤醒，继续执行

## 实例一 ##

场景: 6个同学陆续离开教室后值班同学才可以关门。

**未使用CountDownLatch**

package com.dongguo.juc;
    
    
    /** * @author Dongguo * @date 2021/9/4 0004-8:51 * @description: */
    public class Demo { 
        public static void main(String[] args) { 
    
            for (int i = 1; i <= 6; i++) { 
                new Thread(() -> { 
                    System.out.println(Thread.currentThread().getName() + "离开教室");
                }, "t" + i).start();
            }
            System.out.println("班长锁门");
        }
    }
    运行结果
    班长锁门
    t1离开教室
    t2离开教室
    t3离开教室
    t4离开教室
    t5离开教室
    t6离开教室

发现逻辑上出了问题 ，班长锁门，之后的同学就无法离开教室，所以这个实现是失败的。

**使用CountDownLatch**

package com.dongguo.juc;
    
    import java.util.concurrent.CountDownLatch;
    
    /** * @author Dongguo * @date 2021/9/4 0004-8:51 * @description: */
    public class Demo { 
        public static void main(String[] args) { 
            CountDownLatch countDownLatch = new CountDownLatch(6);
            for (int i = 1; i <= 6; i++) { 
                new Thread(()->{ 
                    countDownLatch.countDown();
                    System.out.println(Thread.currentThread().getName()+"离开教室");
                },"t"+i).start();
            }
            try { 
                countDownLatch.await();
                System.out.println("班长锁门");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }
    运行结果
    t2离开教室
    t3离开教室
    t1离开教室
    t4离开教室
    t5离开教室
    t6离开教室
    班长锁门

可以看到班长等到6位同学全部走光了才会锁门。

我们还可以配合线程池使用，改进如下

package com.dongguo.juc;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /** * @author Dongguo * @date 2021/8/24 0024-16:03 * @description: CountDownLatch */
    public class CountDownLatchDemo2 { 
    
        /** * 6个同学陆续离开教室后值班同学才可以关门 */
        public static void main(String[] args) throws Exception { 
            ExecutorService service = Executors.newFixedThreadPool(4);
            //定义一个数值为6的计数器
            CountDownLatch countDownLatch = new CountDownLatch(6);
            // 创建6个同学
            for (int i = 1; i <= 6; i++) { 
                try { 
                    int count =i;
                    service.submit(() -> { 
                        if (Thread.currentThread().getName().equals("同学6")) { 
                            try { 
                                Thread.sleep(2000);
                            } catch (InterruptedException e) { 
                                e.printStackTrace();
                            }
                        }
                        System.out.println(Thread.currentThread().getName() +"同学"+count+ "离开了"); //计数器减一,不会阻塞
                        countDownLatch.countDown();
                    });
                } catch (
                        Exception e) { 
                    e.printStackTrace();
                }
            }
            //主线程await休息
            System.out.println("主线程睡觉");
            countDownLatch.await();//主线程被唤醒
            //全部离开后自动唤醒主线程
            System.out.println("全部离开了,现在的计数器为" + countDownLatch.getCount());
            service.shutdown();
        }
    }
    运行结果
    主线程睡觉
    pool-1-thread-1同学1离开了
    pool-1-thread-3同学3离开了
    pool-1-thread-3同学5离开了
    pool-1-thread-1同学6离开了
    pool-1-thread-4同学4离开了
    pool-1-thread-2同学2离开了
    全部离开了,现在的计数器为0

## 实例二 ##

模拟10个人匹配LOL进入加载游戏 ，等待10个人全部加载100%后游戏开始

package com.dongguo.juc;
    
    
    import java.util.Arrays;
    import java.util.Random;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    
    /** * @author Dongguo * @date 2021/8/24 0024-16:03 * @description: 模拟10个人匹配LOL进行加载游戏 ，等待10个人全部加载100%后游戏开始 */
    public class CountDownLatchDemo3 { 
    
        public static void main(String[] args) throws Exception { 
            ExecutorService service = Executors.newFixedThreadPool(10);
            Random random = new Random();
            String[] all = new String[10];
            CountDownLatch countDownLatch = new CountDownLatch(10);
            for (int j = 0; j < 10; j++) { //10人
                int k = j;
                service.submit(()->{ 
                    for (int i = 0; i <=100 ; i++) { 
                        try { 
                            //随机睡眠100毫秒以内
                            Thread.sleep(random.nextInt(100));
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                        all[k] = i+"%";
                        System.out.print("\r"+ Arrays.toString(all));
                    }
                    countDownLatch.countDown();
                });
            }
            countDownLatch.await();
            System.out.println("\n游戏开始");
            service.shutdown();
        }
    }
    运行结果
    [100%, 100%, 100%, 100%, 100%, 100%, 100%, 100%, 100%, 100%]
    游戏开始

## 实例三 ##

在分布式系统中，经常会出现多次远程调用不同的服务的场景，

比如查询订单，需要调用订单信息，商品信息，物流信息等服务

package com.dongguo.juc;
    
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /** * @author Dongguo * @date 2021/9/13 0013-16:50 * @description: 模拟三个服务请求 */
    @RestController
    public class CountDownlatchController { 
        /** * 订单服务 * @param id * @return */
        @GetMapping("/order/{id}")
        public Map<String, Object> order(@PathVariable int id) { 
            HashMap<String, Object> map = new HashMap<>();
            map.put("id", id);
            map.put("total", "2300.00");
            sleep(2000);
            return map;
        }
    
        /** * 商品服务 * @param id * @return */
        @GetMapping("/product/{id}")
        public Map<String, Object> product(@PathVariable int id) { 
    
            HashMap<String, Object> map = new HashMap<>();
            if (id == 1) { 
                map.put("name", "小爱音箱");
                map.put("price", 300);
            } else if (id == 2) { 
                map.put("name", "小米手机");
                map.put("price", 2000);
            }
            map.put("id", id);
            sleep(1000);
            return map;
        }
    
        /** * 物流服务 * @param id * @return */
        @GetMapping("/logistics/{id}")
        public Map<String, Object> logistics(@PathVariable int id) { 
            HashMap<String, Object> map = new HashMap<>();
            map.put("id", id);
            map.put("name", "中通快递");
            sleep(2500);
            return map;
        }
    
        private void sleep(int millis) { 
            try { 
                Thread.sleep(millis);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

**未使用CountDownLatch**

package com.dongguo.juc;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.web.client.RestTemplate;
    
    import java.util.Map;
    
    /** * @author Dongguo * @date 2021/9/13 0013-16:49 * @description: */
    @Slf4j(topic = "d.CountDownLatchDemo4")
    public class CountDownLatchDemo4 { 
        public static void main(String[] args) { 
            RestTemplate restTemplate = new RestTemplate();
            log.debug("begin");
            Map<String,Object> response = restTemplate.getForObject("http://localhost:8080/order/{1}", Map.class, 1);
            log.debug("result order:{}",response);
            Map<String,Object> response2 =  restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 1);
            Map<String,Object> response3 =  restTemplate.getForObject("http://localhost:8080/product/{2}", Map.class, 2);
            log.debug("result product:{}",response2,response3);
            Map<String,Object> response4 = restTemplate.getForObject("http://localhost:8080/logistics/{1}", Map.class, 1);
            log.debug("result logistics:{}",response4);
        }
    }
    运行结果
    17:28:40 [main] d.CountDownLatchDemo4 - begin
    17:28:43 [main] d.CountDownLatchDemo4 - result order:{ total=2300.00, id=1}
    17:28:45 [main] d.CountDownLatchDemo4 - result product:{ price=300, name=小爱音箱, id=1}
    17:28:47 [main] d.CountDownLatchDemo4 - result logistics:{ name=中通快递, id=1}

这样串行的请求，效率是非常的，一共花费7S

**使用CountDownLatch+线程池**

package com.dongguo.juc;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.web.client.RestTemplate;
    
    import java.util.Map;
    import java.util.concurrent.*;
    
    /** * @author Dongguo * @date 2021/9/13 0013-16:49 * @description: */
    @Slf4j(topic = "d.CountDownLatchDemo4")
    public class CountDownLatchDemo4 { 
        public static void main(String[] args) { 
            RestTemplate restTemplate = new RestTemplate();
            ExecutorService service = Executors.newCachedThreadPool();
            CountDownLatch countDownLatch = new CountDownLatch(4);
            log.debug("begin");
            service.submit(() -> { 
                Map<String,Object> response =
                        restTemplate.getForObject("http://localhost:8080/order/{1}", Map.class, 1);
                log.debug("result order:{}",response);
                countDownLatch.countDown();
            });
            service.submit(() -> { 
                Map<String,Object> response2 =
                        restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 1);
                log.debug("result product:{}",response2);
                countDownLatch.countDown();
            });
            service.submit(() -> { 
                Map<String,Object> response2 =
                        restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 2);
                log.debug("result product:{}",response2);
                countDownLatch.countDown();
            });
            service.submit(() -> { 
                Map<String,Object> response4 =
                        restTemplate.getForObject("http://localhost:8080/logistics/{1}", Map.class, 1);
                log.debug("result logistics:{}",response4);
                countDownLatch.countDown();
            });
    
            try { 
                countDownLatch.await();
    
                log.debug("执行完毕");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (Exception e) { 
                e.printStackTrace();
            }finally { 
                service.shutdown();
            }
        }
    }
    运行结果
    17:43:28 [main] d.CountDownLatchDemo4 - begin
    17:43:29 [pool-1-thread-2] d.CountDownLatchDemo4 - result product:{ price=300, name=小爱音箱, id=1}
    17:43:29 [pool-1-thread-3] d.CountDownLatchDemo4 - result product:{ price=2000, name=小米手机, id=2}
    17:43:30 [pool-1-thread-1] d.CountDownLatchDemo4 - result order:{ total=2300.00, id=1}
    17:43:31 [pool-1-thread-4] d.CountDownLatchDemo4 - result logistics:{ name=中通快递, id=1}
    17:43:31 [main] d.CountDownLatchDemo4 - 执行完毕

一共花费3s

但是这些打印信息都是在线程池的工作线程中打印的并没有将信息返回给main线程

**使用Future 的get()获得信息**

package com.dongguo.juc;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.web.client.RestTemplate;
    
    import java.util.Map;
    import java.util.concurrent.*;
    
    /** * @author Dongguo * @date 2021/9/13 0013-16:49 * @description: */
    @Slf4j(topic = "d.CountDownLatchDemo4")
    public class CountDownLatchDemo4 { 
        public static void main(String[] args) { 
            RestTemplate restTemplate = new RestTemplate();
            ExecutorService service = Executors.newCachedThreadPool();
            log.debug("begin");
            Future<Map<String,Object>> f1 = service.submit(() -> { 
                Map<String, Object> r =
                        restTemplate.getForObject("http://localhost:8080/order/{1}", Map.class, 1);
                return r;
            });
            Future<Map<String, Object>> f2 = service.submit(() -> { 
                Map<String, Object> r =
                        restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 1);
                return r;
            });
            Future<Map<String, Object>> f3 = service.submit(() -> { 
                Map<String, Object> r =
                        restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 2);
                return r;
            });
            Future<Map<String, Object>> f4 = service.submit(() -> { 
                Map<String, Object> r =
                        restTemplate.getForObject("http://localhost:8080/logistics/{1}", Map.class, 1);
                return r;
            });
    
            try { 
                System.out.println(f1.get());
                System.out.println(f2.get());
                System.out.println(f3.get());
                System.out.println(f4.get());
                log.debug("执行完毕");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            }finally { 
                service.shutdown();
            }
        }
    }
    17:58:40 [main] d.CountDownLatchDemo4 - begin
    { total=2300.00, id=1}
    { price=300, name=小爱音箱, id=1}
    { price=2000, name=小米手机, id=2}
    { name=中通快递, id=1}
    17:58:43 [main] d.CountDownLatchDemo4 - 执行完毕

CountDownLatch是一次性的，初始化后，进行业务操作，计数器归零，则这个实例使命完成，无法服务这个实例。如果想要循环使用那就可以选择CyclicBarrier

# **CyclicBarrier** #

CyclicBarrier看英文单词可以看出大概就是循环阻塞的意思，在使用中CyclicBarrier的构造方法第一个参数是目标障碍数，每次执行CyclicBarrier一次 障碍数会加一，如果达到了目标障碍数，才会执行cyclicBarrier.await()之后的语句。可以将CyclicBarrier 的await()理解为加1操作

场景: 集齐7颗龙珠就可以召唤神龙

package com.dongguo.juc;
    
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    /** * @author Dongguo * @date 2021/9/4 0004-8:51 * @description: */
    public class Demo { 
        //定义神龙召唤需要的龙珠总数
        private final static int NUMBER = 7;
        public static void main(String[] args) { 
            CyclicBarrier cyclicBarrier = new CyclicBarrier(NUMBER, () -> { 
                System.out.println("集齐7颗龙珠召唤神龙");
            });
            for (int i = 1; i <= 7; i++) { 
                int temp = i;
                new Thread(() -> { 
                    try { 
                        System.out.println(Thread.currentThread().getName() + "收集到第" + temp + "颗龙珠");
                        cyclicBarrier.await();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) { 
                        e.printStackTrace();
                    }
                }, "t" + i).start();
            }
        }
    }
    运行结果
    t1收集到第1颗龙珠
    t2收集到第2颗龙珠
    t3收集到第3颗龙珠
    t4收集到第4颗龙珠
    t5收集到第5颗龙珠
    t6收集到第6颗龙珠
    t7收集到第7颗龙珠
    集齐7颗龙珠就可以召唤神龙

# **Semaphore** #

Semaphore的构造方法中传入的第一个参数是最大信号量（可以看成最大线程池），每个信号量初始化为一个最多只能分发一个许可证。使用acquire方法获得许可证，release方法释放许可

Semaphore信号量，用来限制能同时访问共享资源的线程上限

在概念上，信号量维持一组许可证。如果有必要，每个`acquire()`都会阻塞，直到许可证可用，然后才能使用它。每个`release()`添加许可证，潜在地释放阻塞获取方。但是，没有使用实际的许可证对象;`Semaphore`只保留可用数量的计数，并相应地执行。即一个Semaphore维护了一组permits【许可证】。每次调用acquire()方法都会阻塞，直到获取到许可证。每次调用release()方法都会添加一个许可证，也就是释放一个被阻塞的获取者。但是实际上并不存在这个许可证，Semaphore仅仅是记录可用资源的数量，并且做出对应的行为（有资源就获取，没有资源就阻塞）。

**信号量通常用于限制线程数，而不是访问某些（物理或逻辑）资源。**

*  线程池控制的是线程数量，而信号量控制的是并发数量，虽然说看起来一样，但两者还是有区别的。
 *  信号量类似于锁机制，信号量的调用，当达到数量后，线程还是存在的，只是被挂起了而已。而线程池，同时执行的线程数量是固定的，超过了数量的只能等待。

在获得项目之前，每个线程必须从信号量获取许可证，以确保某个项目可用。当线程完成该项目后，它将返回到池中，并将许可证返回到信号量，允许另一个线程获取该项目。请注意，\*\*当调用acquire()时，不会保持同步锁定，因为这将阻止某个项目返回到池中。\*\*信号量封装了限制对池的访问所需的同步，与保持池本身一致性所需的任何同步分开。【**即将限制对池的访问和对池中数据的操作所需要的锁分开**】。

信号量被初始化为一个，并且被使用，使得它只有至多一个允许可用，可以用作互斥锁。这通常被称为二进制信号量，因为它只有两个状态：**一个许可证可用，或零个许可证可用**。当以这种方式使用时，二进制信号量具有属性（与许多Lock实现不同），“锁”可以由除所有者之外的线程释放（因为信号量没有所有权概念）。这在某些专门的上下文中是有用的，例如死锁恢复。

Semaphore(int permits) 创建一个 Semaphore与给定数量的许可证和非公平公平设置。  
    Semaphore(int permits, boolean fair) 创建一个 Semaphore与给定数量的许可证和给定的公平设置。

此类的构造函数可选择接受公平参数。当设置为false时，此类不会保证线程获取许可的顺序。特别是，闯入是允许的，也就是说，一个线程调用acquire()可以提前已经等待线程分配的许可证-在等待线程队列的头部逻辑新的线程将自己【新线程将自己放在等待线程队列的最前面】。当公平设置为真时，信号量保证调用acquire方法的线程被选择以按照它们调用这些方法的顺序获得许可（先进先出; FIFO）【**FIFO的顺序是指是依据到达方法内部的执行点的时间，并不是方法执行的时间。**】。请注意，FIFO排序必须适用于这些方法中的特定内部执行点。因此，一个线程可以在另一个线程之前调用acquire，但是在另一个线程之后到达排序点，并且类似地从方法返回。另请注意，未定义的tryAcquire方法不符合公平性设置，但将采取任何可用的许可证。【**不定时的tryAcquire()方法会任意选取可用的许可证。**】【**非公平锁可以插队获取运行，公平锁按照线程顺序执行。**】

通常，用于控制资源访问的信号量应该被公平地初始化，以确保线程没有被访问资源【**确保没有线程因为长时间获取不到许可证而饿死**】。当使用信号量进行其他类型的同步控制时，非正常排序的吞吐量优势往往超过公平性。

![img][]

场景: 抢车位, 10部汽车3个停车位

package com.dongguo.juc;
    
    
    import java.util.Timer;
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.TimeUnit;
    
    /** * @author Dongguo * @date 2021/9/4 0004-8:51 * @description: */
    public class Demo { 
    
        private final static int NUMBER = 3;
        public static void main(String[] args) { 
            //定义3个停车位
            Semaphore semaphore =new Semaphore(NUMBER);
    
            for (int i = 1; i <=10; i++) { 
                new Thread(()->{ 
                    try { 
                        TimeUnit.MILLISECONDS.sleep(100);//找车位
                        System.out.println(Thread.currentThread().getName() + "找车位ing");
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName() + "汽车停车成功!");
                        TimeUnit.SECONDS.sleep(3);//停车时间
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println(Thread.currentThread().getName() + "离开停车位");
                        semaphore.release();
                    }
                },"t"+i).start();
            }
        }
    }
    运行结果：
    t1找车位ing
    t1汽车停车成功!
    t2找车位ing
    t2汽车停车成功!
    t3找车位ing
    t3汽车停车成功!
    t4找车位ing
    t5找车位ing
    t6找车位ing
    t7找车位ing
    t9找车位ing
    t8找车位ing
    t10找车位ing
    t1离开停车位
    t4汽车停车成功!
    t2离开停车位
    t5汽车停车成功!
    t3离开停车位
    t6汽车停车成功!
    t4离开停车位
    t7汽车停车成功!
    t5离开停车位
    t9汽车停车成功!
    t6离开停车位
    t8汽车停车成功!
    t7离开停车位
    t10汽车停车成功!
    t9离开停车位
    t8离开停车位
    t10离开停车位

[img]: /images/20220828/952e14fb45c14dbbb638c2a65e7829e8.png