Java并发工具之控制并发流程的工具类

忘是亡心i 2023-07-14 10:37 17阅读 0赞

## 1. 什么是控制并发流程？ ##

*  控制并发流程的工具类：帮助我们程序员更容易的让线程之间协作；
 *  让线程之间配合，来满足业务逻辑；
 *  比如：让线程A等待线程B执行完毕后再执行等合作策略；

## 2. 有哪些控制并发流程的工具类？ ##

<table> 
 <thead> 
  <tr> 
   <th>类</th> 
   <th>作用</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Semaphore</td> 
   <td>信号量，可以通过控制“许可证”的数量来保证线程之间的配合，线程只有拿到“许可证”后才能继续运行。相比于其他的同步方法，更加灵活</td> 
  </tr> 
  <tr> 
   <td>CyclicBarrier</td> 
   <td>线程会等待，直到足够多的线程达到了事先规定的数目。一旦达到触发条件，就可以进行下一步的动作。适用于线程之间相互等待处理结果就绪的场景</td> 
  </tr> 
  <tr> 
   <td>Phaser</td> 
   <td>和CyclicBarrier类似，但是计数可变，Java7加入的</td> 
  </tr> 
  <tr> 
   <td>CountDownLatch</td> 
   <td>和CyclicBarrier类似，数量递减到0时，触发动作，但是不可重复使用</td> 
  </tr> 
  <tr> 
   <td>Exchanger</td> 
   <td>让两个线程在合适时交换对象，适用于当两个线程工作在同一个类的不同实例上时，用于交换数据</td> 
  </tr> 
  <tr> 
   <td>Condition</td> 
   <td>可以控制线程的等待和唤醒，是Object.wait()的升级版</td> 
  </tr> 
 </tbody> 
</table>

## 3. CountDownLatch：倒计时唤醒等待的线程 ##

*  中文含义：倒数门闩；
 *  例如：购物拼团，人满发货；
 *  流程：倒数结束之前，一直处于等待状态，直到倒计时结束了，此线程才能继续工作；
 *  CountDownLatch(int count)：仅有一个构造方法，参数count为倒数的值；
 *  await()：调用await方法的线程会被挂起，它会等待直到count值为0才继续执行；
 *  countDown()：将count减一，直到0时，等待的线程才会被唤醒；

两种典型用法：

1.  用法一，一等多：一个线程等待多个线程都执行完毕，再继续自己的工作；

/** * 工厂中，质检，5个工人检查，所有人都认为通过，才通过质检 */
    public class CountDownLatchDemo1 { 
    
        public static void main(String[] args) throws InterruptedException { 
            CountDownLatch countDownLatch = new CountDownLatch(5);
            ExecutorService executorService = Executors.newFixedThreadPool(5);
            for (int i = 0; i < 5; i++) { 
                int finalI = i;
                executorService.submit(new Runnable() { 
                    @Override
                    public void run() { 
                        try { 
                            Thread.sleep(1000);
                            System.out.println("No." + finalI + "完成了检查");
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        } finally { 
                            countDownLatch.countDown();
                        }
                    }
                });
            }
            System.out.println("等待5个人检查完......");
            countDownLatch.await();
            System.out.println("所有人都完成了检查，进入下一个环节");
        }
    }

等待5个人检查完......
    No.0完成了检查
    No.3完成了检查
    No.1完成了检查
    No.4完成了检查
    No.2完成了检查
    所有人都完成了检查，进入下一个环节

1.  用法二，多等一：比如运动员起跑等枪声

/** * 百米竞赛，5名选手都准备好了，等一声令下，所有人同时起跑 */
    public class CountDownLatchDemo2 { 
    
        public static void main(String[] args) throws InterruptedException { 
            CountDownLatch begin = new CountDownLatch(1);
            ExecutorService executorService = Executors.newFixedThreadPool(5);
            for (int i = 0; i < 5; i++) { 
                int finalI = i;
                executorService.submit(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println(finalI + " 准备完毕，等待发令枪");
                        try { 
                            begin.await();
                            System.out.println(finalI + "开始跑步了......");
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            }
            Thread.sleep(1000);
            begin.countDown();
            System.out.println("发令!!!");
        }
    }

0 准备完毕，等待发令枪
    1 准备完毕，等待发令枪
    4 准备完毕，等待发令枪
    2 准备完毕，等待发令枪
    3 准备完毕，等待发令枪
    发令!!!
    0开始跑步了......
    1开始跑步了......
    4开始跑步了......
    3开始跑步了......
    2开始跑步了......

1.  用法三，多等一和一等多：比如运动员起跑等枪声，终点裁判等待所有运动员通过终点

/** * 百米竞赛，5名选手都准备好了，等一声令下，所有人同时起跑 */
    public class CountDownLatchDemo2 { 
    
        public static void main(String[] args) throws InterruptedException { 
            CountDownLatch begin = new CountDownLatch(1);
            CountDownLatch end = new CountDownLatch(5);
            ExecutorService executorService = Executors.newFixedThreadPool(5);
            for (int i = 0; i < 5; i++) { 
                int finalI = i;
                executorService.submit(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println(finalI + " 准备完毕，等待发令枪");
                        try { 
                            begin.await();
                            System.out.println(finalI + "开始跑步了......");
                            Thread.sleep((long) (Math.random() * 10000));
                            System.out.println(finalI + "到达终点");
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        } finally { 
                            end.countDown();
                        }
                    }
                });
            }
            Thread.sleep(1000);
            begin.countDown();
            System.out.println("发令!!!");
            end.await();
            System.out.println("所有人到达终点，比赛结束");
        }
    }

0 准备完毕，等待发令枪
    4 准备完毕，等待发令枪
    1 准备完毕，等待发令枪
    3 准备完毕，等待发令枪
    2 准备完毕，等待发令枪
    发令!!!
    0开始跑步了......
    4开始跑步了......
    1开始跑步了......
    3开始跑步了......
    2开始跑步了......
    0到达终点
    3到达终点
    2到达终点
    1到达终点
    4到达终点
    所有人到达终点，比赛结束

*  不限制多等一，一等多，只要记住await和countDown方法相互配合；
 *  CountDownLatch不能重用，如果需要重新计数，可以考虑使用CyclicBarrier或者创建新的CountDownLatch实例；
 *  CountDownLatch在创建实例的时候，需要传递倒数次数。哪个线程想等待就调用await方法，哪个线程倒数就调用countDown方法，倒数到0的时候，之前等待的线程会继续运行；

## 4. Semaphore信号量：限制同时运行的线程的数量 ##

*  Semaphore用来限制或管理数量有限的资源的使用情况；
 *  信号量的作用是维护一个“许可证”的计数，线程可以获取到许可证，那么信号量剩余的许可证就减一，线程如果释放了许可证，那么信号量剩余的许可证就加一，当信号量所拥有的许可证数量为0，那么下一个还想要获取许可证的线程就需要等待，知道有另外的线程释放了许可证；

信号量使用流程：

1. 初始化Semaphore并指定许可证的数量；
    2. 在需要被限制的代码前加acquire()或者acquireUninterruptibly()；
    3. 执行结束之后，调用release方法来释放许可证；

*  new Semaphore(int permits, boolean fair)：这里可以设置是否要使用公平策略，如果传入true，那么Semaphore会把之前等待的线程放到FIFO的队列里，以便于当有了新的许可证，可以分发给之前等了最长时间的线程；
 *  acquire()能响应中断，acquireUninterruptibly()不能响应中断；
 *  tryAcquire()：看看现在有没有空闲的许可证，如果有的话就获取，如果没有的话，不阻塞而去做别的事；
 *  tryAcquire(timeout)：多了个超时时间；

下面演示基本用法：

public class SemaphoreDemo { 
        static Semaphore semaphore = new Semaphore(3, true);
    
        public static void main(String[] args) { 
            ExecutorService executorService = Executors.newFixedThreadPool(50);
            for (int i = 0; i < 100; i++) { 
                executorService.submit(new Task());
            }
            executorService.shutdown();
        }
    
        static class Task implements Runnable { 
            @Override
            public void run() { 
                try { 
                    semaphore.acquire();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + "拿到了许可证");
                try { 
                    Thread.sleep(2000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + "释放了许可证");
                semaphore.release();
            }
        }
    }

pool-1-thread-2拿到了许可证
    pool-1-thread-3释放了许可证
    pool-1-thread-2释放了许可证
    pool-1-thread-5拿到了许可证
    pool-1-thread-1释放了许可证
    ...........

信号量的特殊用法：一次性获取或释放多个许可证。

比如：TaskA会调用很消耗资源的method1()，而TaskB调用不太消耗资源的method2()，假设我们一共有5个许可证。那么我就可以要求TaskA获取5个许可证才能执行，而TaskB只需要获取到一个许可证就能执行，这样就避免了A和B同时运行的情况，我们可以根据自己的需求合理分配资源。

注意点：

1.获取和释放的数量最好要一致，否则最后许可证数量肯定不够用，导致程序卡死！
    2.注意初始化Semaphore的时候设置公平为true，否则有的线程会插队，导致有些线程一致抢占不到资源，造成线程饥饿；
    3. 获取和释放许可证对线程并无要求，也许A获取了，然后B释放，只要逻辑合理即可；

## 5. Condition接口：升级版的Object.wait() ##

*  当线程1要等待某个条件的时候，就去执行condition.await方法，一旦执行了该方法，线程就进入了阻塞状态；
 *  然后通常会有另外一个线程，假设是线程2，去执行对应的条件，直到这个条件达成的时候，线程2就会去执行condition.signal()方法，这时JVM就会从被阻塞的线程里找，找到那些等待condition的线程，当线程1收到可执行信号的时候，它的线程状态就会变成Runnable状态；
 *  signal()是公平的，只会唤起那个等待时间最长的线程；
 *  signalAll()会唤起所有正在等待的线程；
 *  演示Condition接口的基本用法：

public class ConditionDemo { 
    
        private ReentrantLock lock = new ReentrantLock();
        private Condition condition = lock.newCondition();
    
        public void method1() { 
            lock.lock();
            try { 
                System.out.println("条件不满足，开始await");
                condition.await();
                System.out.println("条件满足了，开始执行后续的任务");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                lock.unlock();
            }
        }
    
        public void method2() { 
            lock.lock();
            try { 
                System.out.println("准备工作完成，开始唤醒其他的线程");
                condition.signal();
            } finally { 
                lock.unlock();
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            ConditionDemo conditionDemo = new ConditionDemo();
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    conditionDemo.method1();
                }
            }).start();
            Thread.sleep(1000);
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    conditionDemo.method2();
                }
            }).start();
        }
    }

条件不满足，开始await
    准备工作完成，开始唤醒其他的线程
    条件满足了，开始执行后续的任务

· 用Condition实现生产者消费者模式：

public class ConditionProducerConsumer { 
    
        public static void main(String[] args) { 
            Storage storage = new Storage();
            new Thread(new Producer(storage)).start();
            new Thread(new Consumer(storage)).start();
        }
    
    }
    
    class Storage { 
    
        Lock lock = new ReentrantLock();
    
        Condition condition = lock.newCondition();
    
        private int MaxSize = 5;
    
        private List<Object> objects = new ArrayList<>();
    
        public void put() { 
            lock.lock();
            try { 
                while (objects.size() == MaxSize) { 
                   condition.await();
                }
                objects.add(new Object());
                System.out.println("生产了一个商品，剩余" + objects.size() + "个");
                condition.signal();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                lock.unlock();
            }
        }
    
        public void take() { 
            lock.lock();
            try { 
                while (objects.size() == 0) { 
                    condition.await();
                }
                objects.remove(0);
                System.out.println("消费了一个商品，剩余" + objects.size() + "个");
                condition.signal();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                lock.unlock();
            }
        }
    }
    
    class Producer implements Runnable { 
    
        private Storage storage;
    
        public Producer(Storage storage) { 
            this.storage = storage;
        }
    
    
        @Override
        public void run() { 
            for (int i = 0; i < 1000; i++) { 
                storage.put();
            }
        }
    }
    
    class Consumer implements Runnable { 
    
        private Storage storage;
    
        public Consumer(Storage storage) { 
            this.storage = storage;
        }
    
        @Override
        public void run() { 
            for (int i = 0; i < 1000; i++) { 
                storage.take();
            }
        }
    }

· 用Condition和Queue实现生产者消费者模式：

public class ConditionQueueProducerConsumer { 
    
        private static int maxSize = 5;
        private static BlockingQueue<Object> queue = new ArrayBlockingQueue<>(maxSize);
    
        public static void main(String[] args) { 
    
            Lock lock = new ReentrantLock();
            Condition notFull = lock.newCondition();
            Condition notEmpty = lock.newCondition();
    
            Thread producer = new Thread(new Runnable() { 
                @Override
                public void run() { 
                    while (true) { 
                        lock.lock();
                        try { 
                            while (queue.size() == maxSize) { 
                                notFull.await();
                            }
                            queue.put(new Object());
                            System.out.println("生产了一个商品，剩余" + queue.size() + "个");
                            notEmpty.signal();
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        } finally { 
                            lock.unlock();
                        }
                    }
                }
            });
            Thread consumer = new Thread(new Runnable() { 
                @Override
                public void run() { 
                    while (true) { 
                        lock.lock();
                        try { 
                            while (queue.size() == 0) { 
                                notEmpty.await();
                            }
                            queue.take();
                            System.out.println("消费了一个商品，剩余" + queue.size() + "个");
                            notFull.signal();
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        } finally { 
                            lock.unlock();
                        }
                    }
                }
            });
    
            producer.start();
            consumer.start();
        }
    }

Condition注意点：

*  实际上，如果说Lock来代替synchronized，那么Condition就是用来代替相对应的Object.wait/notify，所以在用法和性质上，几乎都一样；
 *  await方法会自动释放持有的Lock锁，和Object.wait一样，不需要自己手动释放锁；
 *  wait/notify必须在synchronized同步代码块中使用，而Condition配合Lock使用；
 *  调用await之前，必须首先持有锁，和wait方法性质是一样的，否则会抛出异常；
 *  wait/notify因为必须配合synchronized关键字绑定固定的锁对象，而Condition可以设置多个条件，更加灵活；

## 6. CyclicBarrier循环栅栏：可重用的CountDownLatch ##

*  和CountDownLatch很类似，都能阻塞一组线程；
 *  让大量线程相互配合，并且最后需要统一汇总的时候，可以使用CyclicBarrier构造一个集结点，当某一个线程执行完毕，它就会到集结点等待，知道所有线程都到了集结点，那么该栅栏就会被撤销，所有线程再统一出发，继续执行剩下的任务；
 *  基本用法如下：

public class CyclicBarrierDemo { 
        public static void main(String[] args) { 
            CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("所有线程到达，一起执行");
                }
            });
            ExecutorService executorService = Executors.newFixedThreadPool(5);
            for (int i = 0; i < 5; i++) { 
                executorService.submit(new Task(cyclicBarrier));
            }
            executorService.shutdown();
        }
    
    
        static class Task implements Runnable { 
            CyclicBarrier cyclicBarrier;
    
            public Task(CyclicBarrier cyclicBarrier) { 
                this.cyclicBarrier = cyclicBarrier;
            }
    
            @Override
            public void run() { 
                try { 
                    System.out.println(Thread.currentThread().getName() + "到达集合地点");
                    cyclicBarrier.await();
                } catch (InterruptedException | BrokenBarrierException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

pool-1-thread-1到达集合地点
    pool-1-thread-5到达集合地点
    pool-1-thread-2到达集合地点
    pool-1-thread-3到达集合地点
    pool-1-thread-4到达集合地点
    所有线程到达，一起执行
    pool-1-thread-4到达集合地点
    pool-1-thread-1到达集合地点
    pool-1-thread-2到达集合地点
    pool-1-thread-5到达集合地点
    pool-1-thread-3到达集合地点
    所有线程到达，一起执行

从结果可以看出，CyclicBarrier是可以重用的，规定5个线程到了一起出发，当再来5个线程，还可以出发一波。

CyclicBarrier和CountDownLatch的区别：

1.  CountDownLatch每次用countDown方法来触发倒数事件；
2.  CyclicBarrier是用await来触发倒数事件；
3.  CountDownLatch倒数到0就触发事件，不能重用，而CyclicBarrier可以重复使用，只要达到规定的条件；