> 科技资讯 > [高并发]AQS中的CountDownLatch/Semaphore与CyclicBarrier用法总结

[高并发]AQS中的CountDownLatch/Semaphore与CyclicBarrier用法总结

科技资讯 投稿 5100℃ 0 评论

[高并发]AQS中的CountDownLatch/Semaphore与CyclicBarrier用法总结

CountDownLatch

概述

调用该类await(方法的线程会一直阻塞，直到其他线程调用该类的countDown(方法，使当前计数器的值变为0为止。每次调用该类的countDown(方法，当前计数器的值就会减1。当计数器的值减为0的时候，所有因调用await(方法而处于等待状态的线程就会继续往下执行。这种操作只能出现一次，因为该类中的计数器不能被重置。如果需要一个可以重置计数次数的版本，可以考虑使用CyclicBarrier类。

public boolean await(long timeout, TimeUnit unit

使用场景

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

代码示例

示例代码如下：

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;
    }
}

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

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

概述

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

使用场景

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

概述

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

使用场景

可以用于多线程计算数据，最后合并计算结果的场景

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;
@Slf4j
public 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;
    }
}

编程笔记 » [高并发]AQS中的CountDownLatch/Semaphore与CyclicBarrier用法总结

赞同 (27) or 分享 (0)
游客 发表我的评论   换个身份
取消评论

表情
(0)个小伙伴在吐槽