Java 多线程 : JUC 并发工具原理

愿天堂没有BUG

共 5540字,需浏览 12分钟

 ·

2021-10-16 20:14

一 . 前言

趁着有空 , 赶紧把之前欠的债还上 . 这是多线程一阶段计划的最后一篇 , 后续多线程会转入修订和深入阶段 . 彻底吃透多线程.

二. 工具介绍

之前说 AQS 的时候曾经提到过这几个类 , 这几个类有一些各自的特点 , 很符合特定的场景 , 之前在生产上用的还挺舒服.

我们一般使用的并发工具有四种 :

CyclicBarrier : 放学一起走

  • 允许一组线程互相等待,直到到达某个公共屏障点 (common barrier point)

  • 让一组线程到达一个屏障时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续干活

CountDownLatch : 等人到齐了就触发

  • 在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待

  • 用给定的计数 初始化 CountDownLatch。

  • 由于调用了 countDown() 方法,所以在当前计数到达零之前,await 方法会一直受阻塞。

  • 之后,会释放所有等待的线程,await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。

  • CountDownLatch是通过一个计数器来实现的,当我们在new 一个CountDownLatch对象的时候需要带入该计数器值,该值就表示了线程的数量。

  • 每当一个线程完成自己的任务后,计数器的值就会减1。当计数器的值变为0时,就表示所有的线程均已经完成了任务

Semaphore

  • 信号量Semaphore是一个控制访问多个共享资源的计数器,和CountDownLatch一样,其本质上是一个“共享锁”。

Exchanger

  • 可以在对中对元素进行配对和交换的线程的同步点

  • 每个线程将条目上的某个方法呈现给 exchange 方法,与伙伴线程进行匹配,并且在返回时接收其伙伴的对象 , Exchanger 可能被视为 SynchronousQueue 的双向形式

三 .原理解析

3 .1 CyclicBarrier

作用 :

它允许一组线程互相等待,直到到达某个公共屏障点 (Common Barrier Point)。在涉及一组固定大小的线程的程序中,这些线程必须不时地互相等待,此时 CyclicBarrier 很有用。因为该 Barrier 在释放等待线程后可以重用,所以称它为循环( Cyclic ) 的 屏障( Barrier ) 。

内部原理 :

内部使用重入锁ReentrantLock 和 Condition

构造函数 :

  • CyclicBarrier(int parties):

    • 创建一个新的 CyclicBarrier,它将在给定数量的参与者(线程)处于等待状态时启动,

    • 但它不会在启动 barrier 时执行预定义的操作。

  • CyclicBarrier(int parties, Runnable barrierAction) :

    • 创建一个新的 CyclicBarrier,它将在给定数量的参与者(线程)处于等待状态时启动,

    • 并在启动 barrier 时执行给定的屏障操作,该操作由最后一个进入 barrier 的线程执行。

使用变量 :

  • parties 变量 : 表示拦截线程的总数量。

  • count 变量 : 表示拦截线程的剩余需要数量。

  • barrierAction 变量 : 为 CyclicBarrier 接收的 Runnable 命令,用于在线程到达屏障时,优先执行 barrierAction ,用于处理更加复杂的业务场景。

  • generation 变量 : 表示 CyclicBarrier 的更新换代

// 常用方法 : 
M- await : 等待状态
M- await(long timeout, TimeUnit unit) : 等待超时
M- dowait
- 该方法第一步会试着获取锁
- 如果分代已经损坏,抛出异常
- 如果线程中断,终止CyclicBarrier
- 进来线程 ,--count
- count == 0 表示所有线程均已到位,触发Runnable任务
- 唤醒所有等待线程,并更新generation
> 跳出等待状态的方法
- 最后一个线程到达,即index == 0
- 超出了指定时间(超时等待)
- 其他的某个线程中断当前线程
- 其他的某个线程中断另一个等待的线程
- 其他的某个线程在等待barrier超时
- 其他的某个线程在此barrier调用reset()方法。reset()方法用于将屏障重置为初始状态。
SC- Generation : 描述了 CyclicBarrier 的更新换代。
- 在CyclicBarrier中,同一批线程属于同一代。
- 当有 parties 个线程全部到达 barrier 时,generation 就会被更新换代。
- 其中 broken 属性,标识该当前 CyclicBarrier 是否已经处于中断状态
M- breakBarrier : 终止所有的线程
M- nextGeneration : 更新换代操作
- 1. 唤醒所有线程。
- 2. 重置 count 。
- 3. 重置 generation 。
M- reset : 重置 barrier 到初始化状态
M- getNumberWaiting : 获得等待的线程数
M- 判断 CyclicBarrier 是否处于中断

复制代码

使用案例 :

Gitee CyclicBarrier 使用

问题补充 :


// 问题一 : 拦截的核心
1. 传入总得 Count 数
2. 每次进来都会 --count , 同时判断 count ==0
3. 如果不为 0 ,当前线程就会阻塞

// 问题二 : 涉及源码
private final ReentrantLock lock = new ReentrantLock();
private final Condition trip = lock.newCondition();

复制代码

3.2 CountDownLatch

在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待

用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法,所以在当前计数到达零之前,await 方法会一直受阻塞。之后,会释放所有等待的线程,await 的所有后续调用都将立即返回。这种现象只出现一次, 计数无法被重置。如果需要重置计数,请考虑使用 CyclicBarrier

CountDownLatch是通过一个计数器来实现的,当我们在new 一个CountDownLatch对象的时候需要带入该计数器值,该值就表示了线程的数量。每当一个线程完成自己的任务后,计数器的值就会减1。当计数器的值变为0时,就表示所有的线程均已经完成了任务

// 内部主要方法
> CountDownLatch内部依赖Sync实现,而Sync继承AQS

> sync :
: tryAcquireShared 获取同步状态
: tryReleaseShared 释放同步状态

> await() :
使当前线程在锁存器倒计数至零之前一直等待,除非线程被中断
: sync.acquireSharedInterruptibly(1);
: 内部使用AQS的acquireSharedInterruptibly(int arg)

> getState()
: 获取同步状态,其值等于计数器的值
: 从这里我们可以看到如果计数器值不等于0,则会调用doAcquireSharedInterruptibly(int arg)

> doAcquireSharedInterruptibly
: 自旋方法会尝试一直去获取同步状态

> countDown
: CountDownLatch提供countDown() 方法递减锁存器的计数,如果计数到达零,则释放所有等待的线程
: 内部调用AQS的releaseShared(int arg)方法来释放共享锁同步状态
: tryReleaseShared(int arg)方法被CountDownLatch的内部类Sync重写

复制代码

参考案例

Gitee CountDownLatch 使用

总结

CountDownLatch 内部通过共享锁实现。在创建CountDownLatch实例时,需要传递一个int型的参数:count,该参数为计数器的初始值,也可以理解为该共享锁可以获取的总次数。

当某个线程调用await()方法,程序首先判断count的值是否为0,如果不会0的话则会一直等待直到为0为止 (PS : 可以多个线程都调用 await)

当其他线程调用countDown()方法时,则执行释放共享锁状态,使count值 – 1 (PS :countDown 并不会阻塞)

当在创建CountDownLatch时初始化的count参数,必须要有count线程调用countDown方法才会使计数器count等于0,锁才会释放,前面等待的线程才会继续运行。注意CountDownLatch不能回滚重置

3 .3 Semaphore

基础点

信号量Semaphore是一个控制访问多个共享资源的计数器,和CountDownLatch一样,其本质上是一个“共享锁”

从概念上讲,信号量维护了一个许可集。如有必要,在许可可用前会阻塞每一个 acquire(),然后再获取该许可。每个 release() 添加一个许可,从而可能释放一个正在阻塞的获取者。

Semaphore 通常用于限制可以访问某些资源(物理或逻辑的)的线程数目

当一个线程想要访问某个共享资源时,它必须要先获取Semaphore,当Semaphore >0时,获取该资源并使Semaphore – 1。如果Semaphore值 = 0,则表示全部的共享资源已经被其他线程全部占用,线程必须要等待其他线程释放资源。当线程释放资源时,Semaphore则+1

实现细节

Semaphore提供了两个构造函数:

  • Semaphore(int permits) :创建具有给定的许可数和非公平的公平设置的 Semaphore。

  • Semaphore(int permits, boolean fair) :创建具有给定的许可数和给定的公平设置的 Semaphore。

Semaphore默认选择非公平锁。

当信号量Semaphore = 1 时,它可以当作互斥锁使用。其中0、1就相当于它的状态,当=1时表示其他线程可以获取,当=0时,排他,即其他线程必须要等待。

//------ 信号量获取
> acquire()方法来获取一个许可
: 内部调用AQS的acquireSharedInterruptibly(int arg),该方法以共享模式获取同步状态


> 公平
: 判断该线程是否位于CLH队列的列头
: 获取当前的信号量许可
: 设置“获得acquires个信号量许可之后,剩余的信号量许可数”
: CAS设置信号量
> 非公平
: 不需要判断当前线程是否位于CLH同步队列列头

复制代码

3 .4 Exchanger

可以在对中对元素进行配对和交换的线程的同步点

每个线程将条目上的某个方法呈现给 exchange 方法,与伙伴线程进行匹配,并且在返回时接收其伙伴的对象 , Exchanger 可能被视为 SynchronousQueue 的双向形式

Exchanger,它允许在并发任务之间交换数据 : 当两个线程都到达同步点时,他们交换数据结构,因此第一个线程的数据结构进入到第二个线程中,第二个线程的数据结构进入到第一个线程中

TODO : Exchanger 的源代码比较绕 ,而且这个组件使用场景并不多 , 所以先留个坑 , 以后项目上真的有场景了再实际上分析一下

3.5 并发工具使用

@ github.com/black-ant/c…

补充 :

# CountDownLatch 和 CyclicBarrier 如何理解 ?

  • CyclicBarrier : 小学生去郊游 , 老师下车时统计人数 ,人数到齐了才能一起参观

  • CountDownLatch : 幼儿园老师送孩子(ChildThread)放学 , 走一个记一个数 ,当所有的学生放学后 , 老师(BossThread)下班

// 核心解释 : 
CyclicBarrier 就是一堵墙 , 人数到了所有线程才能一起越过墙
CountDownLatch 只是一个计数器 , 数目到了主线程才能执行

// 其他要点 :
CyclicBarrier 可以重置计数 , CountDownLatch 不可以
复制代码

总结

终于补上了最后一块板 , 后面来真正的深入多线程看看吧 , 争取早日成为多线程大师段位


作者:AntBlack
链接:https://juejin.cn/post/7007360668735602701
来源:稀土掘金
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。



浏览 9
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报