https://at.itzhai.com/itzhai_logo_sm.jpg IT宅

并发编程

并发编程原理和应用
帅旋
关注
充电
IT宅站长，技术博主，架构师，全网id：arthinking。
返回专题首页

图解几个好玩的并发辅助工具类

发布于 2020-03-27 | 更新于 2024-05-16
CountDownLatch
2
CyclicBarrier
2
Semaphore
2

去年问我怎么学Java的那个五年级小学生又来向我问问题了，说Java中线程同步有没有好用的工具类。幸亏没有问我什么算法，瑟瑟发抖。这个我倒是还挺在行的，了解到他喜欢熊出没，于是我就用熊作为主角，给他分享了几个Java并发框架中的辅助工具类。(呼~松了一口气)

image-20200327222907794

看文本片文章，你将了解到：

  1. CountDownLatch的工作原理和使用场景；
  2. CyclicBarrier的工作原理和使用场景；
  3. Semaphore的工作原理和使用场景。

1、闭锁 CountDownLatch

一个同步工具类，允许一个或者多个线程一直等待，直到其他线程的操作都执行完成之后再继续往下执行。

**使用场景：**在一些应用场合中，需要等待某个条件达到要求后才能做后面的事情；同时当线程都完成后也会触发事件，以便进行后面的操作。 这个时候就可以使用CountDownLatch。CountDownLatch最重要的方法是countDown()和await()，前者主要是计数减一，后者是等待计数到0，如果没有到达0，就继续阻塞等待。

方法详细介绍：CountDownLatch的介绍和使用

为了方便理解，这不，我又发挥了以下我的动画绘制功底，写了一个动图：

countdownlatch

如上图，左边三只小熊，可以当成三个线程，每一只撞到栏杆，计数器就减1，这相当于执行了countDown方法；

右边有两只暴走小熊在等待计数器变为0，可以当成两个线程，执行了await方法；

最终左边三只暴走小熊抵达了栏杆处，计数器变为0，唤醒了右边的暴走小熊，暴走小熊就开始动起来了。

1.1、执行原理

CountDownLatch是基于AQS共享模式的使用。

如下图，我们通过给CountDownLatch构造函数传入state的值。

countDown方法本质是释放共享锁，核心实现逻辑是：state>0 && state-1，如果state>0，则state减一，否则执行失败；

await方法本质是获取共享锁，核心实现是：getState()0，如果state0，则表示获取成功，否则线程阻塞进入等待队列；

image-20200317233628586

当state减到0的时候，会唤醒等待队列中的所有线程，尝试继续获取共享锁，这个时候正常是所有线程都能获取成功的。

1.2、使用案例

三个线程共同拉取一块数据，每个线程拉取数据块的一部分，等到所有线程的数据都拉取过来之后，另一个处理线程再开始这个数据块。

下载线程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class Downloader implements Runnable{

	private CountDownLatch latch;

	private String downloaderName;

	/**
	 * 构造函数
	 * @param downLatch 注意, 所有需要协作的线程需要使用同一个闭锁
	 * @param downloaderName
     */
	public Downloader(CountDownLatch downLatch, String downloaderName){
		this.latch = downLatch;
		this.downloaderName = downloaderName;
	}

	public void run() {
		this.download();

		try {
			TimeUnit.SECONDS.sleep(new Random().nextInt(10));
		} catch (InterruptedException e) {
			System.out.println("Thread interrupt status: " + Thread.currentThread().isInterrupted());
		}
		System.out.println(this.downloaderName + "下载完成...");
		this.latch.countDown();
	}

	private void download(){
		System.out.println(this.downloaderName + "正在下载文件...");
	}

}

数据处理线程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class DataProcessor implements Runnable {

	private CountDownLatch latch;

	/**
	 * 构造函数
	 * @param latch 注意, 所有需要协作的线程需要使用同一个闭锁
     */
	public DataProcessor(CountDownLatch latch){
		this.latch = latch;
	}

	public void run() {
		System.out.println("等待下载完数据...");
		try {
			this.latch.await();
		} catch (InterruptedException e) {
			System.out.println("Thread interrupt status: " + Thread.currentThread().isInterrupted());
		}
		System.out.println("数据下载完成, 开始处理数据...");
	}

}

运行代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
ExecutorService executor = Executors.newFixedThreadPool(4);

CountDownLatch latch = new CountDownLatch(3);

Downloader d1 = new Downloader(latch, "下载线程1");
Downloader d2 = new Downloader(latch, "下载线程2");
Downloader d3 = new Downloader(latch, "下载线程3");

DataProcessor processor = new DataProcessor(latch);
executor.execute(d1);
executor.execute(d2);
executor.execute(d3);
executor.execute(processor);
executor.shutdown();

执行结果：

1
2
3
4
5
6
7
8
等待下载完数据...
下载线程1正在下载文件...
下载线程2正在下载文件...
下载线程3正在下载文件...
下载线程3下载完成...
下载线程2下载完成...
下载线程1下载完成...
数据下载完成, 开始处理数据...

1.3、其他说明

类似的，我们也可以使用Thread.join方法实现控制线程执行顺序，但是没有那么灵活。如果我们把任务都丢到线程池里面多线程执行，那么就不能手动的在一个线程里面调用另一个线程的join方法了。

2、栅栏 CyclicBarrier

屏障，或成为栅栏，我们在之前 一文带你彻底理解同步和锁的本质(干货) 一文有讨论到过，主要是达到这种目的：所有线程都准备就绪，就着手下一阶段的工作，否则不能进入下一阶段。

我们还是让上面的小熊来演示一下。

cyclicbarrier

上面5只小熊，准备跑到起跑线，跑到起跑线等待，相当于执行了await方法，等到所有小熊准备就绪之后，然后一起开跑。这就很好的揭示了内存屏障的作用了。

2.1、执行原理

CyclicBarrier是基于ReentrantLock的Condition来实现的。

如下图，栅栏中有两个关键属性：

  • parties：栅栏计数器初始值
  • count：栅栏计数器

其中CyclicBarrier的await()方法封装了对ReentrantLock条件锁的使用，主要处理流程：

  • 获取ReentrantLock锁；
  • count减1，如果此时count为0，那么唤醒等待队列中所有线程，并结束这一轮处理，重置屏障，否则进入下一步；
  • 执行condition.await方法，把当前线程丢到条件队列；
  • 当count减少到0的时候，执行condition.signalAll方法把条件队列中的所有线程节点都移动到等待队列；
  • 最后唤醒同步队列中的线程节点，线程从condition.await阻塞处醒来继续执行：获取ReentrantLock锁，用当前线程节点替换旧的头节点，最终放ReentrantLock锁，继续让线程往下执行(每个线程依次获取、锁释放锁)。如下图：

image-20200319235023688

await()能够响应中断。除此之外，await还提供了带有超时的实现await(long timeout, TimeUnit unit)，以及reset()方法重新开启下一轮，具体大家可以看源码的实现。

2.2、使用案例

下面的案例模拟了赛跑，只有当所有运动员都在起跑线上准备好了，才允许他们开跑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
public class CyclicBarrierTest {

	public static void main(String[] args) {
		CyclicBarrier barrier = new CyclicBarrier(3);
		ExecutorService executor = Executors.newFixedThreadPool(3);
		executor.submit(new Runner(barrier, "1号选手"));
		executor.submit(new Runner(barrier, "2号选手"));
		executor.submit(new Runner(barrier, "3号选手"));
		barrier.reset();
		executor.submit(new Runner(barrier, "4号选手"));
		executor.submit(new Runner(barrier, "5号选手"));
		executor.submit(new Runner(barrier, "6号选手"));
		executor.shutdown();
	}
}

class Runner implements Runnable {

	/**
	 * 一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)
	 */
	private CyclicBarrier barrier;

	private String name;

	public Runner(CyclicBarrier barrier, String name) {
		super();
		this.barrier = barrier;
		this.name = name;
	}

	@Override
	public void run() {
		try {
			Thread.sleep(1000 * (new Random()).nextInt(8));
			System.out.println(name + " 准备好了...");
			// barrier的await方法，在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。
			barrier.await();
		} catch (InterruptedException | BrokenBarrierException e) {
			e.printStackTrace();
		}
		System.out.println(name + " 起跑！");
		for (int i = 0; i < 10; i++) {
			System.out.println(this.name + "正在跑步" + i);
		}
	}
}

2.3、其他说明

CountDownLatch是若干个线程等待另外n个线程完成某件事之后才能执行；而CyclicBarrier是若干个线程互相等待，只有等到所有线程都执行了await只会，这若干个线程才可以继续往下执行。

3、信号量 Semaphore

信号量通过一组许可证来控制对共享资源的访问。

如果需要，可以用acquire()方法获取许可，如果许可为0，那么会进行阻塞，通过使用release()方法释放许可，把许可归还给Semaphore，归还之后，阻塞的线程优惠醒来尝试获取许可。

Semaphore提供给了若干个api对应不同的功能：

  • Semaphore(int permits)：非公平模式创建；
  • Semaphore(int permits, boolean fair)：可以指定是否公平模式创建；
  • acquire()：尝试获取1个许可，如果没有许可则阻塞，可以被中断停止等待；
  • acquire(int permits)：跟上一个方法类型，尝试获取permits个许可；
  • acquireUninterruptibly()：尝试获取一个许可，不可中断；
  • acquireUninterruptibly(int permits)：尝试获取permits个许可，不可中断；
  • tryAcquire()：尝试获取一个许可，获取不到则直接返回失败；
  • tryAcquire(int permits)：尝试获取permits个许可，获取不到则直接返回失败；
  • tryAcquire(int permits, long timeout, TimeUnit unit)：尝试在timeout时间内获取permits个许可，超时则返回false，可被中断；
  • tryAcquire(long timeout, TimeUnit unit)：尝试在timeout时间内获取1个许可，超时则返回false，可被中断；
  • release()：释放一个许可；
  • release(int permits)：释放n个许可；

下面演示基于公平锁的Semaphore，获取锁使用acquireUninterruptibly()：

semaphore

这里设置的许可为2，可以发现，同一时刻最多只能有两个线程获得许可。

3.1、执行原理

Semaphore的执行原理相对来说比较简单。下面描述了可中断非公平的信号量实现原理，ASQ中的state值就相当于许可的数量：

  • 执行acquire的时候，会尝试让state - acquires，如果发现许可足够，则进行cas更新，扣减许可，否则线程进入等待队列；
  • 执行release的时候，state + releases，把许可加回去。

image-20200321181406717

3.2、使用案例

下面演示了使用semaphore实现限流的机制，模拟20个客户端线程尝试执行业务逻辑，同一时刻最多只有5个线程能够并发的执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// 线程池
ExecutorService exec = Executors.newCachedThreadPool();
// 只能5个线程同时访问
final Semaphore semp = new Semaphore(5);
// 模拟20个客户端访问
for (int index = 0; index < 20; index++) {
  final int NO = index;
  Runnable run = () -> {
    try {
      // 获取许可
      if(semp.tryAcquire()) {
        System.out.println("线程获得许可: " + NO);
        Thread.sleep((long) (Math.random() * 10000));
        // 访问完后，释放
        semp.release();
      } else {
        System.out.println("达到并发上限,请求失败,请稍后再试");
      }

    } catch (InterruptedException e) {
      System.out.println("执行异常");
    }
  };
  exec.execute(run);
}
// 退出线程池
exec.shutdown();

注意，这里使用的是tryAcquire失败之后直接返回，线程不会进入AQS等待队列。

图解BlockingQueue阻塞队列
如何优雅的挂起线程

本文作者： 帅旋

本文链接： https://www.itzhai.com/columns/cpj/concurrent-helper-classes.html

版权声明： 版权归作者所有，未经许可不得转载，侵权必究！联系作者请加公众号。

×
IT宅

关注公众号及时获取网站内容更新。

请帅旋喝一杯咖啡

咖啡=电量，给帅旋充杯咖啡，他会满电写代码！

IT宅

关注公众号及时获取网站内容更新。