29 理解 并发 countDownLatch
image.png
image.png
image.png
image.png
image.png
分析lock 锁的原理 基于javaAQS 类封装 ，并发队列
lock原理 基于javaAQS类封装 在获取锁的时候AQS类中有一个状态state+1，当前线程不断重入 的时候都会不断+1，在释放锁的时候state-1。
最终state为0 的时候 该锁没有被任何线程获取到。没有抢到锁的线程会存在一个双向的链表中。
公平锁与非公平锁的时候多了一个判断
(!hasQueuedPredecessors() &
阻塞和唤醒用的apilocksupport,为了这个效率只会唤醒阻塞队列中head节点.next 线程。
AQS中为什么头结点是为空的，没有绑定任何线程
答案： 头结点就是我们获取到锁的线程记录，为了gc回收所有的头结点内容变为null.
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//自旋唤醒线程
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//线程进入阻塞
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
synchronized 底层采用c++写
锁池主要存放：
等待池： 当前线程如果调用wait方法 ，单向链表
lock 基于AQS：
同步阻塞队列：主要存放的就是我们没有抢到锁的线程 双向链表 Condition阻塞队列： 主要存放当前线程如果调用await 方法，单向链表存放await.
Await 方法：
1， 存放COndition单向链表的最后位置
2，释放锁 锁的状态为0.
signal 唤醒原理：
1，采用cas 将wait 状态由-2 变为0
2， 将该线程追加到双向链表同步队列中 状态-1 开始竞争锁
3， 调用释放锁的代码的时候，唤醒t1线程
condition.signal() 唤醒当前线程
condition.signalAll() 根据头结点遍历唤醒所有节点的线程。
package com.taotao.metithread;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
*@author tom
*Date 2020/7/26 0026 8:16
*condition 用法
*/
public class Test031 {
private static Lock lock=new ReentrantLock();
private static Condition condition=lock.newCondition();
public static void main(String[] args) {
new Thread(()->{
System.out.println("1");
try {
lock.lock();
//当前子线程释放锁，同时变为阻塞状态
condition.await();
}catch (Exception e){
}
System.out.println(2);
lock.unlock();
},"t1").start();
try {
Thread.sleep(3000);
}catch (Exception e){
}
System.out.println("3");
lock.lock();
//主线程唤醒t1
condition.signal();
lock.unlock();
}
}
单向节点
/**
* Adds a new waiter to wait queue.
* @return its new wait node
*/
private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
Condition源码分析
Condition是一个接口，其提供的就两个核心方法，await和signal方法。分别对应着Object的wait和notify方法。调用Object对象的这两个方法，需要在同步代码块里面，即必须先获取到锁才能执行这两个方法。同理，Condition调用这两个方法，也必须先获取到锁
1.等待队列：用于存放在lock锁中调用await方法,当前线程会变为阻塞状态，
同时会释放锁 单向链表存放
3.同步队列：用于存放没有竞争到锁，采用双向链表存放。
等待池
Condition源码解读
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 将当前节点添加到最后一个节点
Node node = addConditionWaiter();
//释放锁的状态
long savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
// 如果当前线程为-2 则当前线程变为阻塞状态
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
//重新获取锁
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
//获取单向链表，
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
AQS 双向链表 当前锁已经持有，存放没有抢到锁线程
Condition 单向链表 调用await 释放锁，当前线程阻塞
CountDownLatch 信号量 灵活唤醒线程 计数器
Semamhore 实现接口的限流 信号量
底层基于aqs 封装
Lock 公平锁，非公平
package com.taotao.metithread;
import java.util.concurrent.CountDownLatch;
/**
*@author tom
*Date 2020/7/28 0028 7:43
*手写countdownlatch
*/
public class Test36 {
private static Object lock = new Object();
public static void main(String[] args) throws InterruptedException {
// new Thread(()->{
// System.out.println(1);
// synchronized (lock){
// try{
// //当前线程释放锁，当前线程变为阻塞
// lock.wait();
// }catch (Exception e){
//
// }
// System.out.println(2);
// }
// }).start();
// Thread.sleep(3000);
// System.out.println(3);
// synchronized (lock){
// try {
// lock.notify();
// }catch (Exception e){
//
// }
// }
//将aqs状态设置为2
CountDownLatch countDownLatch = new CountDownLatch(2);
new Thread(() -> {
System.out.println(1);
try {
//当前线程释放锁，当前线程变为阻塞
countDownLatch.await();
} catch (Exception e) {
}
System.out.println(2);
}).start();
Thread.sleep(3000);
System.out.println(3);
countDownLatch.countDown();
countDownLatch.countDown();
//aqs状态为-1只有aqs状态为0的时候才会唤醒子线程
}
}
手写countdownlatch思路：
1，设置aqs 类中的状态为2；
2，调用await方法， 让当前线程变为阻塞；
3，调用countDown 方法的时候 状态-1，如果状态=0的情况下唤醒刚刚阻塞的线程。
package com.taotao.metithread;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
/**
*@author tom
*Date 2020/7/28 0028 8:04
*手写countdownlatch
*/
public class MayiktCountDownLatch {
private Sync sync;
public MayiktCountDownLatch(int count) {
sync = new Sync(count);
}
//当前线程变为阻塞
public void await() {
sync.acquireShared(1);
}
public void countDown() {
sync.releaseShared(1);
}
class Sync extends AbstractQueuedSynchronizer {
public Sync(int count) {
setState(count);
}
/**
* 如果返回值小于0 的情况下则让当前线程放入到aqs双向链表中
* @param arg
* @return
*/
@Override
protected int tryAcquireShared(int arg) {
//如果aqs的状态>0，则让当前线程放入到aqs双向链表中，返回《0 -1；
return getState() == 0 ? 1 : -1;
}
//如果当期线程返回true 则唤醒刚才阻塞的线程
@Override
protected boolean tryReleaseShared(int arg) {
for (; ; ) {
int oldState = getState();
if (oldState == 0) {
return false;
}
int newState = (oldState) - arg;
if (compareAndSetState(oldState, newState)) {
return newState == 0;
}
}
}
}
public static void main(String[] args) throws InterruptedException {
MayiktCountDownLatch mayiktCountDownLatch=new MayiktCountDownLatch(2);
new Thread(() -> {
System.out.println(Thread.currentThread().getName()+":"+1);
try {
//当前线程释放锁，当前线程变为阻塞
mayiktCountDownLatch.await();
} catch (Exception e) {
}
System.out.println(2);
}).start();
Thread.sleep(3000);
System.out.println(3);
mayiktCountDownLatch.countDown();
mayiktCountDownLatch.countDown();
//aqs状态为-1只有aqs状态为0的时候才会唤醒子线程
}
}
seampoer信号量的使用：
package com.taotao.metithread;
import java.util.concurrent.Semaphore;
/**
*@author tom
*Date 2020/7/28 0028 8:55
*Sempore() 信号量限流
*/
public class Test039 {
public static void main(String[] args) throws InterruptedException {
//设置aqs状态为5 只能限制5个线程执行代码
Semaphore semaphore= new Semaphore(5);
for (int i = 0; i <= 20; i++) {
new Thread(()->{
try {
//aqs状态减1如果状态为0的情况下会编外阻塞
semaphore.acquire();
System.out.println(Thread.currentThread().getName()+"获取门票：");
//aqs状态加1同时唤醒阻塞的状态
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
}
semaphore: 作用对我们的接口实现限流，信号量
1，初始化aqs 为5；
2 当每次调用acquire方法的时候，信号量减1，如果状态为0的时候当前线程阻塞。
3 调用release 方法放时候 ，状态+1同时唤醒刚刚的线程。
package com.taotao.metithread;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
/**
*@author tom
*Date 2020/7/28 0028 9:17
*手写Semapore 信号量
*/
public class MeiteSemaphore {
private Sync sync;
private MeiteSemaphore(int count) {
sync = new Sync(count);
}
public void acquire() {
sync.acquireShared(1);
}
public void release(){
sync.releaseShared(1);
}
class Sync extends AbstractQueuedSynchronizer {
public Sync(int count) {
setState(count);
}
@Override
protected int tryAcquireShared(int arg) {
for (; ; ) {
int oldState = getState();
int newState = oldState - arg;
if (compareAndSetState(oldState, newState)) {
return newState < 0 ? -1 : 1;
}
}
}
@Override
protected boolean tryReleaseShared(int arg) {
for (; ; ) {
int oldState = getState();
int newState = oldState+arg;
if(newState<oldState){
throw new Error(" not found");
}
if (compareAndSetState(oldState, newState)) {
return true;
}
}
}
}
public static void main(String[] args) throws InterruptedException {
//设置aqs状态为5 只能限制5个线程执行代码
MeiteSemaphore semaphore= new MeiteSemaphore(5);
for (int i = 0; i <= 20; i++) {
new Thread(()->{
//aqs状态减1如果状态为0的情况下会编外阻塞
semaphore.acquire();
System.out.println(Thread.currentThread().getName()+"获取门票：");
//aqs状态加1同时唤醒阻塞的状态
semaphore.release();
}).start();
}
}
}
Semaphore信号量底层原理
Semaphore用于限制可以访问某些资源（物理或逻辑的）的线程数目，他维护了一个许可证集合，有多少资源需要限制就维护多少许可证集合，假如这里有N个资源，那就对应于N个许可证，同一时刻也只能有N个线程访问。一个线程获取许可证就调用acquire方法，用完了释放资源就调用release方法。
可以简单理解为Semaphore信号量可以实现对接口限流，底层是基于aqs实现
Semaphore信号量底层原理
Semaphore用于限制可以访问某些资源（物理或逻辑的）的线程数目，他维护了一个许可证集合，有多少资源需要限制就维护多少许可证集合，假如这里有N个资源，那就对应于N个许可证，同一时刻也只能有N个线程访问。一个线程获取许可证就调用acquire方法，用完了释放资源就调用release方法。
可以简单理解为Semaphore信号量可以实现对接口限流，底层是基于aqs实现
CountDownLatch
CountDownLatch（计数器）、CyclicBarrier（回环屏障）、信号量（Semaphore）
CountDownLatch源码分析
CountDownLatch是一种java.util.concurrent包下一个同步工具类，它允许一个或多个线程等待直到在其他线程中一组操作执行完成。 和join方法非常类似
CountDownLatch底层是基于AQS实现的
CountDownLatch countDownLatch=new CountDownLatch(2) AQS的state状态为2
调用countDownLatch.countDown();方法的时候状态-1 当AQS状态state为0的情况下，则唤醒正在等待的线程。
CountDownLatch与Join的区别
Join底层是基于wait方法实现，而CountDownLatch底层是基于AQS实现。
