并发编程（三）：并发模拟

  我们要如何模拟一个高并发的环境来检查我们的代码呢？一般常用的方式有三种：利用测试工具ab，利用jmeter，代码模拟。本篇博客着重要说的是代码模拟的实现方式。

  在开始写代码之前我们首先了解一下J.U.C中特别重要的两个工具类：CountDownLatch、Semaphore

  CountDownLatch（闭锁）

  CountDownLatch是一个同步工具类，它允许一个或多个线程一直等待，直到其他线程的操作执行完后再执行。CountDownLatch是通过一个计数器来实现的，当计数器的值为0时，在闭锁上等待的线程就可以恢复执行任务

    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }

  从此类的构造方法可以看出，构造此类需要传入count，这个值只能被设置一次，而且这个计数值不能被重置

  countDown（）方法：每调用一次这个方法，构造函数中的count值减1

  await（）方法：调用此方法的线程会在这个方法上阻塞，直到count值为0才继续执行自己的任务

 CountDownLatch-demo1

@Slf4j
public class CountDownLatchExample1 {

    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) {
                    log.error("exception",e);
                }finally {
                    countDownLatch.countDown();
                }

            });
        }

        countDownLatch.await();
        log.info("finish");
        //线程池不再使用后关闭
        exec.shutdown();

    }

    private static void test(int threadNum)throws Exception {
        Thread.sleep(100);
        log.info("{}", threadNum);
        Thread.sleep(100);
    }
}

输出如下：

  此demo1证明了count为0后发生阻塞的线程（调用await（）处）才会继续执行

  CountDownLatch-demo2

@Slf4j
public class CountDownLatchExample2 {

    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) {
                    log.error("exception",e);
                }finally {
                    countDownLatch.countDown();
                }

            });
        }
        //统计的时间为test方法执行的时间 ，如下所示为只等待10毫秒就继续执行
        countDownLatch.await(10, TimeUnit.MILLISECONDS);
        log.info("finish");
        //线程池不再使用后关闭，不是直接销毁，会让当前已有的线程执行完然后销毁
        exec.shutdown();

    }

    private static void test(int threadNum)throws Exception {
        Thread.sleep(100);
        log.info("{}", threadNum);
//        Thread.sleep(100);
    }
}

输出如下：

  demo2表明await()方法支持设置阻塞时间，如果等待时间查过设置的阻塞时间，则发生阻塞的线程继续向下执行（可参考await（）的重载）

  Semaphore（信号量）

  Semaphore管理一系列许可证，Semaphore的acquire请求许可证，获得许可证可继续执行，否则发生阻塞，Semaphore的release方法释放许可证。下面我们来看一下Semaphore的构造方法

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

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

  从构造方法可以看出，两个构造方法都必须提供许可证的数量（可理解为同时并发执行的线程数），不同的是第二个构造方法还需要指定是公平模式还是非公平模式，默认为非公平模式。公平模式是指调用acquire的顺序就是获取许可证的顺序，遵循FIFO；非公平模式是抢占式的，可能一个新的获取线程恰好在许可证释放的时候得到了许可证，而它前面还有在等待许可证的线程

Semaphore-demo1

@Slf4j
public class SemaphoreExample1 {

    private final static int threadCount = 20;

    public static void main(String[] args) throws Exception{
        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(()->{
                try {
                    semaphore.acquire(); //获取一个许可
                    test(threadNum);
                    semaphore.release();  //释放一个许可
                } catch (Exception e) {
                    log.error("exception",e);
                }
            });
        }

        //线程池不再使用后关闭
        exec.shutdown();

    }

    private static void test(int threadNum)throws Exception {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

输出如下：

参考demo1的输出可发现同一秒内，只有三个线程在并发执行（信号量为3每次只获取一个许可）

Semaphore-demo2

@Slf4j
public class SemaphoreExample2 {

    private final static int threadCount = 20;

    public static void main(String[] args) throws Exception{
        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(()->{
                try {
                    //并发数为3，并且一下拿走三个许可，已经没有别的许可可以放出来了，所以一个个执行
                    semaphore.acquire(3); //获取多个许可
                    test(threadNum);
                    semaphore.release(3);  //释放多个许可
                } catch (Exception e) {
                    log.error("exception",e);
                }
            });
        }

        //线程池不再使用后关闭
        exec.shutdown();

    }

    private static void test(int threadNum)throws Exception {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

输出如下：

  参考demo2输出可发现，同一秒内只有一个线程在执行（信号量为3，每次获取三个许可），可参考acquire（）方法的重载

Semaphore-demo3

@Slf4j
public class SemaphoreExample3 {

    private final static int threadCount = 20;

    public static void main(String[] args) throws Exception{
        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(()->{
                try {
                    //20个请求在同一时间内都会去尝试获取许可，只有三个线程获得了许可
                    if (semaphore.tryAcquire()) {
                        test(threadNum);
                        semaphore.release();  //释放一个许可
                    }
                } catch (Exception e) {
                    log.error("exception",e);
                }
            });
        }

        //线程池不再使用后关闭
        exec.shutdown();

    }

    private static void test(int threadNum)throws Exception {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

输出如下：

  注意上图为demo3完整输出，本次代码中使用的是tryAcquire（）而不是acquire（）方法，tryAcquire（）表示线程会尝试获取许可，如果无法获取许可则此请求被丢弃，信号量为3，每次获取一个许可，所以只有三个线程被执行。

 Semaphore-demo4

@Slf4j
public class SemaphoreExample4 {

    private final static int threadCount = 20;

    public static void main(String[] args) throws Exception{
        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(()->{
                try {
                    //只等待5s，没有执行完的线程则丢弃
                    if (semaphore.tryAcquire(5000,TimeUnit.MILLISECONDS)) {
                        test(threadNum);
                        semaphore.release();  //释放一个许可
                    }
                } catch (Exception e) {
                    log.error("exception",e);
                }
            });
        }

        //线程池不再使用后关闭
        exec.shutdown();
    }
    private static void test(int threadNum)throws Exception {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

输出如下：

  参考demo4输出可发现，20个线程只执行了13个，由代码可看出，在5秒内得到许可的线程得到执行（信号量为3，每次获取一个许可），没有得到许可的线程被丢弃（可参考tryAcquire的重载）

  高并发-demo

@Slf4j
public class CountExample1 {

    //请求总数
    public static int clientTotal = 5000;
    //同时并发执行的线程数
    public static int threadTotal = 200;

    public static int count = 0;

    private  static void add() {
        count++;
    }

    public static void main(String[] args)throws Exception {

        //定义线程池
        ExecutorService executorService = Executors.newCachedThreadPool();
        //定义信号量
        final Semaphore semaphore = new Semaphore(threadTotal);
        //定义计数器闭锁
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);

        for (int i = 0; i < clientTotal; i++) {
            executorService.execute(()->{
                try {
                    semaphore.acquire();
                    add();
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception",e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("count:{}",count);

    }
}

输出如下：

 有个以上基础，我们可知此demo模拟了5000个请求，并发数为200的情况，根据输出结果可知发生了线程安全问题。