目錄

1.Semaphore介紹

1.1 限流操作實(shí)戰(zhàn)

1.2 Semaphore源碼解析

?2. countDownLatch介紹

2.1 多個(gè)線程等待實(shí)戰(zhàn)

?2.2 主線程等待

2.3 源碼分析

1.Semaphore介紹

semaphorer(信號(hào)量)，是一個(gè)基于AQS框架實(shí)現(xiàn)的工具類，也是操作系統(tǒng)PV操作在java中的實(shí)現(xiàn)。通過發(fā)放許可來控制線程，只有拿到許可的線程才能執(zhí)行代碼，常用于限流操作。

PV操作是一種操作系統(tǒng)實(shí)現(xiàn)進(jìn)程互斥與同步的有效方法：P表示通過，V表示釋放。

P操作：S-1=X，如果X>=0線程執(zhí)行，如果小于0放入等待隊(duì)列中。

V操作：S+1=X，如果X>=0線程執(zhí)行，如果小于0從等待隊(duì)列釋放一個(gè)等待線程。

構(gòu)造器（默認(rèn)非公平鎖，也有公平鎖實(shí)現(xiàn)）:

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

常用的方法：

acquire()獲取許可拿到共享鎖

tryacquire()嘗試獲取鎖，獲取到鎖返回true，沒有返回false

release()釋放許可

1.1 限流操作實(shí)戰(zhàn)

模擬限流操作

1. 5個(gè)線程發(fā)送3個(gè)許可。

2. 10個(gè)核心線程發(fā)送5個(gè)許可。

import java.util.concurrent.Semaphore;

/**
 * 限流操作
 */
public class SemaphoreTest {

    private static Semaphore semaphore = new Semaphore(3);//發(fā)放三個(gè)許可

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

        //執(zhí)行5個(gè)線程
        for(int i=0;i<5;i++){
            new Thread(()->{
                try{
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName()+"開始購票");
                    Thread.sleep(5000);
                    System.out.println(Thread.currentThread().getName()+"購票成功");
                }catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    semaphore.release();
                }
            }).start();

        }

    }
}

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 使用線程池-限流操作
 */
public class SemaphoreTest2 {

    //定義一個(gè)5個(gè)許可的信號(hào)量
    private static Semaphore semaphore = new Semaphore(5);

    //定義一個(gè)10個(gè)線程的線程池
    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10,20,60,
            TimeUnit.SECONDS,new LinkedBlockingQueue<>(200));

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

        //每100ms發(fā)送一次請(qǐng)求
        for(;;){
            Thread.sleep(100);
            threadPoolExecutor.execute(()->exec());
            System.out.println("-------------------------------------");
        }
    }

    private static void exec(){
        //模擬限流場(chǎng)景
        try{
            semaphore.acquire(1);
            //暫停2秒
            Thread.sleep(2000);
            System.out.println("執(zhí)行業(yè)務(wù)邏輯");
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            semaphore.release(1);
        }
    }
}

1.2 Semaphore源碼解析

關(guān)注點(diǎn)：

1.加解鎖邏輯(共享鎖)實(shí)現(xiàn)

2.競(jìng)爭鎖失敗入隊(duì)操作阻塞和喚醒同步隊(duì)列中等待線程邏輯

服了這代碼了，寫的狗屁不通

?

?

?2. countDownLatch介紹

?countDownLatch(閉鎖)是一個(gè)同步協(xié)助類，用于一個(gè)或多個(gè)線程等待，相當(dāng)于一個(gè)倒計(jì)時(shí)計(jì)數(shù)器，需要當(dāng)計(jì)數(shù)器為0時(shí)才釋放等待線程執(zhí)行，等待線程會(huì)一次性全部返回，計(jì)數(shù)器不會(huì)重置，需要重置需要使用CyclicBarrier

?常用方法：

public void await() 讓線程等待
public boolean await(long timeout, TimeUnit unit) 讓線程等待一段時(shí)間，沒有置為0執(zhí)行業(yè)務(wù)
countDown() count減1

countDownLatch比join方法更加靈活，join方法底層在不停檢測(cè)線程狀態(tài)是否存活。

countDownLatch與CyclicBarrier區(qū)別：

1.countDownLatch只能使用一次，CyclicBarrier可以通過reset方法重復(fù)使用

2.countDownLatch是通過AQS的共享鎖實(shí)現(xiàn)的，CyclicBarrier是通過reentrantlock的獨(dú)占鎖和condition實(shí)現(xiàn)線程喚醒。

2.1 多個(gè)線程等待實(shí)戰(zhàn)

定義一個(gè)計(jì)數(shù)器為3的閉鎖，線程掛起，等待2秒計(jì)數(shù)器減1，直到最后一個(gè)線程減到0跑線程

import java.util.concurrent.CountDownLatch;

public class CountDownLatchTest {

    //定義一個(gè)計(jì)數(shù)器為3的閉鎖
    private static CountDownLatch countDownLatch = new CountDownLatch(3);

    public static void main(String[] args) throws InterruptedException {
        for (int i =0 ;i<5;i++){
            //阻塞線程
            new Thread(()->{
                try{
                    countDownLatch.await();
                    System.out.println(Thread.currentThread().getName()+"執(zhí)行業(yè)務(wù)");
                }catch(Exception e){
                    e.printStackTrace();
                }
            }).start();

            //等待2秒計(jì)數(shù)器減1
            Thread.sleep(2000);
            countDownLatch.countDown();
        }

    }
}

?2.2 主線程等待

import java.util.concurrent.CountDownLatch;

public class CountDownLatchTest2 {

    public static void main(String[] args) throws InterruptedException {
        //1. 定義一個(gè)計(jì)數(shù)器為3的線程
        CountDownLatch countDownLatch = new CountDownLatch(3);

        //2.讓線程計(jì)數(shù)器減1
        for(int i=0;i<3;i++){
            new Thread(()->{
                try{
                    Thread.sleep(1000);
                    countDownLatch.countDown();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }).start();
        }

        //3.主線程阻塞，當(dāng)計(jì)數(shù)器為0時(shí)執(zhí)行業(yè)務(wù)
        countDownLatch.await();
        System.out.println("執(zhí)行業(yè)務(wù)");
    }
}

2.3 源碼分析

關(guān)注點(diǎn)：

加解鎖邏輯

1. countDownLatch.await()加鎖：默認(rèn)寫死1永遠(yuǎn)waitStatus為-1走阻塞線程邏輯

2. countDownLatch.countDown()解鎖：cas操作減1，直到為0執(zhí)行喚醒線程邏輯?

解決synchronized兩個(gè)線程導(dǎo)致的死鎖問題：使用wait方法釋放鎖的形式解決，線上只能kill了

?