常用的并发工具类

CountDownLatch概念

CountDownLatch是一个同步工具类，用来协调多个线程之间的同步，或者说起到线程之间的通信（而不是用作互斥的作用）。

CountDownLatch能够使一个线程在等待另外一些线程完成各自工作之后，再继续执行。使用一个计数器进行实现。计数器初始值为线程的数量。当每一个线程完成自己任务后，计数器的值就会减一。当计数器的值为0时，表示所有的线程都已经完成一些任务，然后在CountDownLatch上等待的线程就可以恢复执行接下来的任务。

CountDownLatch的用法

CountDownLatch典型用法：

1、某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为new CountDownLatch(n)，每当一个任务线程执行完毕，就将计数器减1 countdownLatch.countDown()，当计数器的值变为0时，在CountDownLatch上await()的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。

CountDownLatch典型用法：

2、实现多个线程开始执行任务的最大并行性。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的CountDownLatch(1)，将其计算器初始化为1，多个线程在开始执行任务前首先countdownlatch.await()，当主线程调用countDown()时，计数器变为0，多个线程同时被唤醒。

CountDownLatch的不足

CountDownLatch是一次性的，计算器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当CountDownLatch使用完毕后，它不能再次被使用。

CountDownLatch（倒计时计算器）使用说明

方法说明

public void countDown()

　　递减锁存器的计数，如果计数到达零，则释放所有等待的线程。如果当前计数大于零，则将计数减少.

public boolean await(long timeout,TimeUnit unit) throws InterruptedException

　　使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断或超出了指定的等待时间。如果当前计数为零，则此方法立刻返回true值。

　　如果当前计数大于零，则出于线程调度目的，将禁用当前线程，且在发生以下三种情况之一前，该线程将一直出于休眠状态：

　　由于调用countDown()方法，计数到达零；或者其他某个线程中断当前线程；或者已超出指定的等待时间。

如果计数到达零，则该方法返回true值。
如果当前线程，在进入此方法时已经设置了该线程的中断状态；或者在等待时被中断，则抛出InterruptedException，并且清除当前线程的已中断状态。
如果超出了指定的等待时间，则返回值为false。如果该时间小于等于零，则该方法根本不会等待。

参数：

　　timeout-要等待的最长时间

　　unit-timeout 参数的时间单位

　　如果计数到达零，则返回true；如果在计数到达零之前超过了等待时间，则返回false

抛出：

　　InterruptedException-如果当前线程在等待时被中断

例子1：

　　主线程等待子线程执行完成在执行

package com.example.demo.CountDownLatchDemo;


import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * 主线程等待子线程执行完成再执行
 */
public class CountdownLatchTest1 {
    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(3);
        final CountDownLatch latch = new CountDownLatch(3);
        for (int i = 0; i < 3; i++) {
            Runnable runnable = new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("子线程" + Thread.currentThread().getName() + "开始执行");
                        Thread.sleep((long) (Math.random() * 10000));
                        System.out.println("子线程"+Thread.currentThread().getName()+"执行完成");
                        latch.countDown();//当前线程调用此方法，则计数减一
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            service.execute(runnable);
        }


        try {
            System.out.println("主线程"+Thread.currentThread().getName()+"等待子线程执行完成...");
            latch.await();//阻塞当前线程，直到计数器的值为0
            System.out.println("主线程"+Thread.currentThread().getName()+"开始执行...");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

例子2：

　　百米赛跑，4名运动员选手到达场地等待裁判口令，裁判一声口令，选手听到后同时起跑，当所有选手到达终点，裁判进行汇总排名

package com.example.demo.CountDownLatchDemo;


import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class CountdownLatchTest2 {
    public static void main(String[] args) {
        ExecutorService service = Executors.newCachedThreadPool();
        final CountDownLatch cdOrder = new CountDownLatch(1);
        final CountDownLatch cdAnswer = new CountDownLatch(4);
        for (int i = 0; i < 4; i++) {
            Runnable runnable = new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("选手" + Thread.currentThread().getName() + "正在等待裁判发布口令");
                        cdOrder.await();
                        System.out.println("选手" + Thread.currentThread().getName() + "已接受裁判口令");
                        Thread.sleep((long) (Math.random() * 10000));
                        System.out.println("选手" + Thread.currentThread().getName() + "到达终点");
                        cdAnswer.countDown();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            service.execute(runnable);
        }
        try {
            Thread.sleep((long) (Math.random() * 10000));
            System.out.println("裁判"+Thread.currentThread().getName()+"即将发布口令");
            cdOrder.countDown();
            System.out.println("裁判"+Thread.currentThread().getName()+"已发送口令，正在等待所有选手到达终点");
            cdAnswer.await();
            System.out.println("所有选手都到达终点");
            System.out.println("裁判"+Thread.currentThread().getName()+"汇总成绩排名");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        service.shutdown();
    }
}

CountDownLatch和CyclicBarrier

CyclicBarrier在用法上其实跟CountDownLatch十分相似，但是前者功能更加强大。

CountDownLatch举例：

CountDownLatch countDownLatch=new CountDownLatch(n);

当程序多次执行countDownLatch.countDown();导致计数器n=0时，阻塞的线程都将同时被唤醒。但是此时的n已经是等于0了，也就是说这个计数器就是一次性的。

CyclicBarrier举例：

CyclicBarrier cyclicBarrier = new CyclicBarrier(n);

CyclicBarrier也能实现countDownLatch的功能，并且它的计数器n是可以被重置的，也就是说n=0线程被唤醒后，n又能重新回到原有值。

CyclicBarrier实现CountDownLatch的功能

public class CycleBarrierTest {
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
        // 当计数器为0时，立即执行
        @Override
        public void run() {
            System.out.println("汇总线程：" + Thread.currentThread().getName() + " 任务合并。");
        }
    });


    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(2);


        // 将线程A添加到线程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("线程A：" + Thread.currentThread().getName() + "执行任务。");
                    System.out.println("线程A：到达屏障点");
                    cyclicBarrier.await();
                    System.out.println("线程A：退出屏障点");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });


        // 将线程B添加到线程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("线程B：" + Thread.currentThread().getName() + "执行任务。");
                    System.out.println("线程B：到达屏障点");
                    cyclicBarrier.await();
                    System.out.println("线程B：退出屏障点");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });


        // 关闭线程池
        executorService.shutdown();
    }
}

执行结果：

线程A：pool-1-thread-1执行任务。
线程A：到达屏障点
线程B：pool-1-thread-2执行任务。
线程B：到达屏障点
汇总线程：pool-1-thread-2 任务合并。
线程B：退出屏障点
线程A：退出屏障点

结果分析：

CyclicBarrier指定了计数器为2，线程A执行cyclicBarrier.await();到达屏障点阻塞，此时计数器等于2-1=1，线程B也执行cyclicBarrier.await();到达屏障点阻塞，此时计数器等于0，CyclicBarrier里的任务会立即执行，执行完成后，线程A和B也被同时唤醒继续执行。

CyclicBarrier相对CountDownLatch的优化

public class CycleBarrierTest2 {
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);


    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(2);


        // 将线程A添加到线程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("线程A：" + Thread.currentThread().getName() + "执行第1步。");
                    cyclicBarrier.await();
                    System.out.println("线程A：" + Thread.currentThread().getName() + "执行第2步。");
                    cyclicBarrier.await();
                    System.out.println("线程A：" + Thread.currentThread().getName() + "执行第3步。");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });


        // 将线程B添加到线程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("线程B：" + Thread.currentThread().getName() + "执行第1步。");
                    cyclicBarrier.await();
                    System.out.println("线程B：" + Thread.currentThread().getName() + "执行第2步。");
                    cyclicBarrier.await();
                    System.out.println("线程B：" + Thread.currentThread().getName() + "执行第3步。");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });


        // 关闭线程池
        executorService.shutdown();
    }
}

执行结果：

线程A：pool-1-thread-1执行第1步。
线程B：pool-1-thread-2执行第1步。
线程B：pool-1-thread-2执行第2步。
线程A：pool-1-thread-1执行第2步。
线程A：pool-1-thread-1执行第3步。
线程B：pool-1-thread-2执行第3步。

结果分析：

CyclicBarrier指定了计数器为2，线程A和线程B都分别执行了2次cyclicBarrier.await();这样导致的效果就是：每个线程分阶段完成任务，等所有线程都完成了第1步，然后才能接着执行第2步，等所有线程都完成了第2步，才能执行第3步。

图解：

CyclicBarrier的特性就在于它的计数器是可以重置的，这也就有了上图那种多屏障的情况，虽然第一次调用await使得计数器等于0屏障1失效，但是后续如果继续调用await，屏障还能继续使用，这也就是计数器重置的好处。而CountDownLatch第一次屏障失效后，就结束了。

Semaphore，信号量，一般用于控制同时访问资源的线程数量。可以认为Synchronized代表的是一把锁，那么Semaphore就是多把锁。

常用方法

public class Semaphore implements java.io.Serializable {
    //构造方法，传入令牌数，默认实例化一个非公平锁
    public Semaphore(int permits);
    //获取一个令牌，在获取成功之前，以及被其他线程中断之前，当前线程会被阻塞
    public void acquire() throws InterruptedException;
    //获取一个令牌，在获取成功之前，当前线程会被阻塞（中断被忽略）
    public void acquireUninterruptibly() ;
    //尝试获取令牌，立即返回获取成功与否，不阻塞当前线程
    public boolean tryAcquire();
    //释放一个令牌
    public void release();
    //返回当前可用的令牌数
    public int availablePermits();
}

现在有这样的一个例子：

某卫生间只有3个坑位，把坑前面的挡门理解为令牌，因此这里有3个令牌，现在模拟5个人抢坑位的场景。

package com.xue.testSemaphore;
 
import java.util.concurrent.Semaphore;
 
public class Main {
    public static void main(String[] args) {
        //最多支持3个人同时蹲坑
        Semaphore semaphore = new Semaphore(3);
        //5个人来抢坑位
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + "已经在蹲坑");
                    //模拟蹲坑时长
                    Thread.sleep((long) (Math.random() * 10 * 1000));
                    //离开坑位
                    System.out.println(Thread.currentThread().getName() + "即将离开坑位");
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, i + "号").start();
        }
    }
}

输出如下：

首先0、1、2号已经抢完了所有的坑位，3与4号只能在外面等候，对的他们没排队（默认实例化了一个非公平锁）。2号出来后，3号才能进去。接着0号出来，4号才能进去。

这个例子虽然有点俗，这确实能让人印象深刻呀。

原理解析

类图

Semaphore有2个内部类，FairSync与NonfairSync，它们都继承自Sync，Sync又继承自AQS。可以看的出来，Semaphore与CountDownLatch的结构类似，都需要借助于AQS。

构造方法

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}


public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

默认实例化了一个非公平锁，当然也可以进行指定。这里的permits最终会传到AQS的state变量中，代表当前可用的令牌数。

acquire()

获取一个令牌，获取到线程可以继续执行，否则将会被阻塞。

public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
}

调用了AQS中的模版方法

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        //尝试获取arg个令牌，该方法返回可用令牌数-需求数，如果小于0，则进行阻塞
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
}

其中tryAcquireShared()由具体的子类(AQS的子类Sync的子类NonfailSync)进行实现

protected int tryAcquireShared(int acquires) {
            return nonfairTryAcquireShared(acquires);
        }

这里又调用了父类Sync的方法

final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
}

Exchanger

public class ExchangerDemo {
 
    static  Exchanger<String> exchanger = new Exchanger <>();
 
    public static void main(String[] args) {
        String t1String = "T1";
        String t2String = "T2";
 
        new Thread(()->{
            try {
                System.out.println("T1交换之前的值：" + t1String);
                String exchange = exchanger.exchange(t1String);
                System.out.println("T1交换之后的值：" + exchange);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"T1").start();
 
        new Thread(()->{
            try {
                System.out.println("T2交换之前的值：" + t2String);
                String exchange = exchanger.exchange(t2String);
                System.out.println("T2交换之后的值：" + exchange);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"T2").start();
    }
}

案例输出：T1交换之前的值：T1

T2交换之前的值：T2

T2交换之后的值：T1

T1交换之后的值：T2

丛输出结果可以看出T1、T2线程实现了数据交换，当T1线程调用exchanger.exchange(t1String)的时候就表示它把t1String拿出来跟其他线程做交换，当有T2调用exchanger.exchange(t2String)的时候就表示把t2String拿出来跟其他线程做交互，当有两条线程要做交换的时候，Exchanger就满足了交换的条件，就会进行值的交换，并且把交换后的值返回。

使用注意点：

1、需要交换值的线程必须是成对的出现。

2、当有两条线程将值交换完成后，Exchanger实例可重复使用，其他线程也可以使用此Exchanger实例来进行数据交换。

Exchanger的实现原理：

Exchanger交换值的原理是使用了ThreadLocal来实现的，至于线程阻塞的部分是使用java中的Usafe类park()/unpark()来实现的。

常用的并发工具类

CountDownLatch和CyclicBarrier

CyclicBarrier实现CountDownLatch的功能

CyclicBarrier相对CountDownLatch的优化

常用方法

原理解析

类图

构造方法

acquire()

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method