29 理解并发 countDownLatch

作者: 滔滔逐浪 | 来源:发表于2020-07-25 07:39 被阅读0次

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实现。

29 理解并发 countDownLatch

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

29 理解 并发 countDownLatch

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

29 理解并发 countDownLatch