高并发之——浅谈AQS中的CountDownLatch、Semaphore与CyclicBarrie

时间 2021-08-12

标签 java 数据库多线程并发 ide 高并发线程 code 资源栏目 Java 繁體版

原文原文链接

CountDownLatch

概述

同步辅助类，经过它能够阻塞当前线程。也就是说，可以实现一个线程或者多个线程一直等待，直到其余线程执行的操做完成。使用一个给定的计数器进行初始化，该计数器的操做是原子操做，即同时只能有一个线程操做该计数器。java

调用该类await()方法的线程会一直阻塞，直到其余线程调用该类的countDown()方法，使当前计数器的值变为0为止。每次调用该类的countDown()方法，当前计数器的值就会减1。当计数器的值减为0的时候，全部因调用await()方法而处于等待状态的线程就会继续往下执行。这种操做只能出现一次，由于该类中的计数器不能被重置。若是须要一个能够重置计数次数的版本，能够考虑使用CyclicBarrier类。数据库

CountDownLatch支持给定时间的等待，超过必定的时间再也不等待，使用时只须要在countDown()方法中传入须要等待的时间便可。此时，countDown()方法的方法签名以下：多线程

public boolean await(long timeout, TimeUnit unit)

使用场景

在某些业务场景中，程序执行须要等待某个条件完成后才能继续执行后续的操做。典型的应用为并行计算：当某个处理的运算量很大时，能够将该运算任务拆分红多个子任务，等待全部的子任务都完成以后，父任务再拿到全部子任务的运算结果进行汇总。并发

代码示例

调用ExecutorService类的shutdown()方法，并不会第一时间内把全部线程所有都销毁掉，而是让当前已有的线程所有执行完，以后，再把线程池销毁掉。ide

示例代码以下：高并发

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CountDownLatchExample {
    private static final int threadCount = 200;

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

        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 (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        log.info("finish");
        exec.shutdown();
    }

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

支持给定时间等待的示例代码以下：ui

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@Slf4j
public class CountDownLatchExample {
    private static final int threadCount = 200;

    public static void main(String[] args) throws InterruptedException {
        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 (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await(10, TimeUnit.MICROSECONDS);
        log.info("finish");
        exec.shutdown();
    }

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

Semaphore

概述

控制同一时间并发线程的数目。可以完成对于信号量的控制，能够控制某个资源可被同时访问的个数。线程

提供了两个核心方法——acquire()方法和release()方法。acquire()方法表示获取一个许可，若是没有则等待，release()方法则是在操做完成后释放对应的许可。Semaphore维护了当前访问的个数，经过提供同步机制来控制同时访问的个数。Semaphore能够实现有限大小的链表。code

使用场景

Semaphore经常使用于仅能提供有限访问的资源，好比：数据库链接数。资源

代码示例

每次获取并释放一个许可，示例代码以下：

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
    private static final int threadCount = 200;

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

        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 (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        exec.shutdown();
    }

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

每次获取并释放多个许可，示例代码以下：

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
    private static final int threadCount = 200;

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

        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(3);  //获取多个许可
                    test(threadNum);
                    semaphore.release(3);  //释放多个许可
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        log.info("finish");
        exec.shutdown();
    }

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

假设有这样一个场景，并发过高了，即便使用Semaphore进行控制，处理起来也比较棘手。假设系统当前容许的最高并发数是3，超过3后就须要丢弃，使用Semaphore也能实现这样的场景，示例代码以下：

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
    private static final int threadCount = 200;

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

        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的源码
                    if (semaphore.tryAcquire()) { 
                        test(threadNum);
                        semaphore.release();  //释放一个许可
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        log.info("finish");
        exec.shutdown();
    }
    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

CyclicBarrier

概述

是一个同步辅助类，容许一组线程相互等待，直到到达某个公共的屏障点，经过它能够完成多个线程之间相互等待，只有当每一个线程都准备就绪后，才能各自继续往下执行后面的操做。

与CountDownLatch有类似的地方，都是使用计数器实现，当某个线程调用了CyclicBarrier的await()方法后，该线程就进入了等待状态，并且计数器执行加1操做，当计数器的值达到了设置的初始值，调用await()方法进入等待状态的线程会被唤醒，继续执行各自后续的操做。CyclicBarrier在释放等待线程后能够重用，因此，CyclicBarrier又被称为循环屏障。

使用场景

能够用于多线程计算数据，最后合并计算结果的场景。

CyclicBarrier与CountDownLatch的区别

CountDownLatch的计数器只能使用一次，而CyclicBarrier的计数器可使用reset()方法进行重置，而且能够循环使用
CountDownLatch主要实现1个或n个线程须要等待其余线程完成某项操做以后，才能继续往下执行，描述的是1个或n个线程等待其余线程的关系。而CyclicBarrier主要实现了多个线程之间相互等待，直到全部的线程都知足了条件以后，才能继续执行后续的操做，描述的是各个线程内部相互等待的关系。
CyclicBarrier可以处理更复杂的场景，若是计算发生错误，能够重置计数器让线程从新执行一次。
CyclicBarrier中提供了不少有用的方法，好比：能够经过getNumberWaiting()方法获取阻塞的线程数量，经过isBroken()方法判断阻塞的线程是否被中断。

代码示例

示例代码以下。

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {

    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);

    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++){
            final int threadNum = i;
            Thread.sleep(1000);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
    private static void race(int threadNum) throws Exception{
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        cyclicBarrier.await();
        log.info("{} continue", threadNum);
    }
}

设置等待超时示例代码以下：

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
@Slf4j
public class CyclicBarrierExample {

    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);

    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++){
            final int threadNum = i;
            Thread.sleep(1000);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
    private static void race(int threadNum) throws Exception{
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        try{
            cyclicBarrier.await(2000, TimeUnit.MILLISECONDS);
        }catch (BrokenBarrierException | TimeoutException e){
            log.warn("BarrierException", e);
        }
        log.info("{} continue", threadNum);
    }
}

在声明CyclicBarrier的时候，还能够指定一个Runnable，当线程达到屏障的时候，能够优先执行Runnable中的方法。

示例代码以下：

package io.binghe.concurrency.example.aqs;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CyclicBarrier;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;@Slf4jpublic class CyclicBarrierExample {    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> {        log.info("callback is running");    });    public static void main(String[] args) throws Exception {        ExecutorService executorService = Executors.newCachedThreadPool();        for (int i = 0; i < 10; i++){            final int threadNum = i;            Thread.sleep(1000);            executorService.execute(() -> {                try {                    race(threadNum);                } catch (Exception e) {                    e.printStackTrace();                }            });        }        executorService.shutdown();    }    private static void race(int threadNum) throws Exception{        Thread.sleep(1000);        log.info("{} is ready", threadNum);        cyclicBarrier.await();        log.info("{} continue", threadNum);    }}