并发编程06-Tools&CountDownLatch&Semaphore原理与应用（CyclicBarrier等）

蔚落 2023-02-28 11:21 5阅读 0赞

**JUC提供的Tools之五：**

**1、** **Semaphore**

**2、** **CountDownLatch**

**3、** **CyclicBarrier**

**4、** **Executors**

**5、** **Exchanger**

![20200721105800676.png][]

**一、Semaphore **

Semaphore 字面意思是信号量的意思，它的作用是控制访问特定资源的线程数目。Semaphore跟springcloud里Hystrix豪猪一样，可以做限流。

**怎么使用 Semaphore？**

**1.1、构造方法**

public Semaphore(int permits)

public Semaphore(int permits, boolean fair)

*  permits 表示许可线程的数量
 *  fair 表示公平性，如果这个设为 true 的话，下次执行的线程会是等待最久的线程。

**1.2、重要方法**

public void acquire() throws InterruptedException

public void release()

tryAcquire（long timeout, TimeUnit unit）

*  acquire() 表示阻塞并获取许可
 *  release() 表示释放许可

**1.3、基本使用**

**1.3.1 需求场景**

资源访问，服务限流。

**1.3.2代码实现**

package my.mark.mybaibaoxiang.juc;

import java.util.concurrent.Semaphore;

/\*\*

\* @author wxming.

\* @description :**可用于流量控制，限制最大的并发访问数。控制访问特定资源的线程数目**

\*/

public class SemaphoreSample \{

public static void main(String\[\] args) \{

Semaphore semaphore = new Semaphore(2);//信号量2，同时只能有2个线程执行。

for (int i = 0; i < 5; i++) \{

new Thread(new Task(semaphore, "yangguo+" + i)).start();

static class Task extends Thread \{

Semaphore semaphore;

public Task(Semaphore semaphore, String tname) \{

this.semaphore = semaphore;

this.setName(tname);

public void run() \{

try \{

semaphore.acquire();

System.out.println(Thread.currentThread().getName() + ":aquire() at time:" + System.currentTimeMillis());

Thread.sleep(5000);

semaphore.release();

System.out.println(Thread.currentThread().getName() + ":release() at time:" + System.currentTimeMillis());

\} catch (InterruptedException e) \{

e.printStackTrace();

**打印结果：**

Thread-1:aquire() at time:1594884535792

Thread-3:aquire() at time:1594884535792

Thread-3:release() at time:1594884540792

Thread-7:aquire() at time:1594884540792

Thread-1:release() at time:1594884540792

Thread-5:aquire() at time:1594884540792

Thread-5:release() at time:1594884545792

Thread-7:release() at time:1594884545792

Thread-9:aquire() at time:1594884545792

Thread-9:release() at time:1594884550793

**从打印结果可以看出，一次只有两个线程执行 acquire()，只有线程进行 release() 方法后**

**才会有别的线程执行 acquire()。**

**二、CountDownLatch使用及应用场景** **例子**

**2.1、CountDownLatch是什么？**

CountDownLatch这个类能够 **使一个线程等待其他线程完成各自的工作后再执行** 。例

如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有的框架服务之后再执行。

**2.2、CountDownLatch如何工作？**

CountDownLatch是通过一个 **计数器**来实现的， **计数器的初始值为线程的数量**。 **每当**

**一个线程完成了自己的任务后，计数器的值就会减1**。当计数器值到达0时，它表示所有的

线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。

**2.3、API**

**CountDownLatch** .countDown()//计数器减一

**CountDownLatch** .await();//开始等待。

**2.4、CountDownLatch应用场景例子**

比如陪媳妇去看病。

医院里边排队的人很多，如果一个人的话，要先看大夫，看完大夫再去排队交钱取药。

现在我们是双核，可以同时做这两个事（多线程）。

假设看大夫花3秒钟，排队交费取药花5秒钟。我们同时搞的话，5秒钟我们就能完成，然后

一起回家（回到主线程）。

**代码如下：**

**主线程**

package my.mark.mybaibaoxiang.juc.countdownlaunch;

import java.util.concurrent.CountDownLatch;

/\*\*

\* @author wxming.

\* @description ：CountDownLatch使用示例：两个人一起去医院，一个人去买药，一个人去看医生。两个人都完事后一起回。

\*/

public class CountDownLaunchSample \{

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

long now = System.currentTimeMillis();

CountDownLatch countDownLatch = new CountDownLatch(2);//初始化计数器数量2。

new Thread(new SeeDoctorTask(countDownLatch)).start();//看医生任务开始。

new Thread(new QueueTask(countDownLatch)).start();//买药任务开始。

//等待线程池中的2个任务执行完毕，否则一直等

countDownLatch.await();

System.out.println("over，回家 cost:" + (System.currentTimeMillis() - now));

**子线程1：**

package my.mark.mybaibaoxiang.juc.countdownlaunch;

import java.util.concurrent.CountDownLatch;

/\*\*

\* @author wxming.

\* @description ： 看医生任务

\*/

public class SeeDoctorTask implements Runnable \{

private CountDownLatch countDownLatch;

public SeeDoctorTask(CountDownLatch countDownLatch) \{

this.countDownLatch = countDownLatch;

public void run() \{

try \{

System.out.println("开始看医生");

Thread.sleep(2000);

System.out.println("看医生结束，准备离开病房");

\} catch (InterruptedException e) \{

e.printStackTrace();

\} finally \{

if (countDownLatch != null)

countDownLatch.countDown();//计数器数量减1.

**子线程2：**

package my.mark.mybaibaoxiang.juc.countdownlaunch;

import java.util.concurrent.CountDownLatch;

/\*\*

\* @author wxming.

\* @description ：排队买药任务

\*/

public class QueueTask implements Runnable \{

private CountDownLatch countDownLatch;

public QueueTask(CountDownLatch countDownLatch) \{

this.countDownLatch = countDownLatch;

public void run() \{

try \{

System.out.println("开始在医院药房排队买药....");

Thread.sleep(5000);

System.out.println("排队成功，可以开始缴费买药");

\} catch (InterruptedException e) \{

e.printStackTrace();

\} finally \{

if (countDownLatch != null)

countDownLatch.countDown();//计数器数量减1.

**三、** **CyclicBarrier**

栅栏屏障，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程

到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。

CyclicBarrier 默认的构造方法是 CyclicBarrier （ int parties ），其参数表示屏障拦截的线

程数量，每个线程调用 await 方法告 CyclicBarrier 我已经到达了屏障，然后当前线程被阻塞。

**3.1、API**

**cyclicBarrier** .await();

**3.2、** **应用场景**

可以用于多线程计算数据，最后合并计算结果的场景。例如，用一个 Excel 保存了用户

所有银行流水，每个 Sheet 保存一个账户近一年的每笔银行流水，现在需要统计用户的日均

银行流水，先用多线程处理每个 sheet 里的银行流水，都执行完之后，得到每个 sheet 的日 均银行流水，最后，再用 barrierAction 用这些线程的计算结果，计算出整个 Excel 的日 均银行流水。

**3.3、示例代码**

package my.mark.mybaibaoxiang.juc;

import java.util.concurrent.CyclicBarrier;

/\*\*

\* @author wxming.

\* @description :栅栏屏障，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程 到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。

\* CyclicBarrier默认的构造方法是CyclicBarrier（int parties），其参数表示屏障拦截的线 程数量，每个线程调用await方法告CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。

\*/

public class CyclicBarrierTest implements Runnable \{

private CyclicBarrier cyclicBarrier;

private int index;

public CyclicBarrierTest(CyclicBarrier cyclicBarrier, int index) \{

this.cyclicBarrier = cyclicBarrier;

this.index = index;

public void run() \{

try \{

System.out.println("index: " + index);

index--;

cyclicBarrier.await();

\} catch (Exception e) \{

e.printStackTrace();

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

CyclicBarrier cyclicBarrier = new CyclicBarrier(11, new Runnable() \{

public void run() \{

System.out.println("所有特工到达屏障，准备开始执行秘密任务");

System.out.println("验证“全部到达屏障”和这个哪个先执行");

\});

for (int i = 0; i < 10; i++) \{

new Thread(new CyclicBarrierTest(cyclicBarrier, i)).start();

System.out.println("子线程全部启动了");

cyclicBarrier.await();//11个，当10个子线程包括这行这个11个都到达屏障，才会执行初始化CyclicBarrier时创建的那个“所有特工到达屏障”的线程，TODO 执行完之后才会继续往下执行。给初始化线程打多个断点之后的日志如下：

/\*\*

\* 子线程全部启动了

index: 0

index: 2

index: 1

index: 3

index: 4

index: 5

index: 6

index: 8

index: 7

index: 9

所有特工到达屏障，准备开始执行秘密任务

验证“全部到达屏障”和这个哪个先执行

全部到达屏障....

\*/

System.out.println("全部到达屏障....");

**四、** **Executors**

主要用来创建线程池，代理了线程池的创建，使得你的创建入口参数变得简单。

**重要方法**

*  newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
 *  newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
 *  newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。
 *  newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

package my.mark.mybaibaoxiang.juc;

import java.util.concurrent.Executors;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

/\*\*

\* @author wxming.

\* @description :创建线程池

\*/

public class ExecutorsTest \{

public static void main(String\[\] args) \{

ScheduledExecutorService es = Executors.newScheduledThreadPool(1);

//延迟三秒执行

es.schedule(new Runnable() \{

public void run() \{

System.out.println("我在跑......");

\}, 3, TimeUnit.SECONDS);

es.shutdown();

**五、** **Exchanger**

当一个线程运行到exchange()方法时会阻塞，另一个线程运行到exchange()时，二者 交换数据，然后执行后面的程序。

应用场景很少。

**代码示例：**

package my.mark.mybaibaoxiang.juc;

import java.util.concurrent.Exchanger;

/\*\*

\* @author wxming.

\* @description :当一个线程运行到exchange()方法时会阻塞，另一个线程运行到exchange()时，二者 交换数据，然后执行后面的程序。

\*/

public class ExchangerTest \{

public static void main(String \[\]args) \{

final Exchanger<Integer> exchanger = new Exchanger<Integer>();

for(int i = 0 ; i < 10 ; i++) \{

final Integer num = i;

new Thread() \{

public void run() \{

System.out.println("我是线程：Thread\_" + this.getName() + "我的数据是：" + num);

try \{

Integer exchangeNum = exchanger.exchange(num);

Thread.sleep(1000);

System.out.println("我是线程：Thread\_" + this.getName() + "我原先的数据为：" + num + " , 交换后的数据为：" + exchangeNum);

\} catch (InterruptedException e) \{

e.printStackTrace();

\}.start();

日志：

我是线程：Thread\_Thread-0我的数据是：0

我是线程：Thread\_Thread-1我的数据是：1

我是线程：Thread\_Thread-2我的数据是：2

我是线程：Thread\_Thread-3我的数据是：3

我是线程：Thread\_Thread-4我的数据是：4

我是线程：Thread\_Thread-5我的数据是：5

我是线程：Thread\_Thread-6我的数据是：6

我是线程：Thread\_Thread-7我的数据是：7

我是线程：Thread\_Thread-8我的数据是：8

我是线程：Thread\_Thread-9我的数据是：9

我是线程：Thread\_Thread-1我原先的数据为：1 , 交换后的数据为：0

我是线程：Thread\_Thread-2我原先的数据为：2 , 交换后的数据为：3

我是线程：Thread\_Thread-7我原先的数据为：7 , 交换后的数据为：6

我是线程：Thread\_Thread-5我原先的数据为：5 , 交换后的数据为：4

我是线程：Thread\_Thread-3我原先的数据为：3 , 交换后的数据为：2

我是线程：Thread\_Thread-0我原先的数据为：0 , 交换后的数据为：1

我是线程：Thread\_Thread-6我原先的数据为：6 , 交换后的数据为：7

我是线程：Thread\_Thread-4我原先的数据为：4 , 交换后的数据为：5

我是线程：Thread\_Thread-9我原先的数据为：9 , 交换后的数据为：8

我是线程：Thread\_Thread-8我原先的数据为：8 , 交换后的数据为：9

**两两交换。**

[20200721105800676.png]: /images/20230209/fd9d2266cb9c44a9b6404bc5363e2ecf.png

并发编程06-Tools&CountDownLatch&Semaphore原理与应用（CyclicBarrier等）

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Java并发工具：CountDownLatch与CyclicBarrier应用

相关 Java并发编程：CountDownLatch与CyclicBarrier的应用实例

相关 Java并发编程：Fork/Join、CyclicBarrier等问题

相关 Java并发编程：CountDownLatch、CyclicBarrier等工具使用示例

相关 Java并发工具：CountDownLatch、CyclicBarrier等实战应用案例

相关 Java并发编程：CountDownLatch与CyclicBarrier应用实例

相关 Java 并发编程——CyclicBarrier

相关并发编程06-Tools&CountDownLatch&Semaphore原理与应用（CyclicBarrier等）

相关并发编程--volatile应用与原理

相关并发编程--CyclicBarrier屏障