手寫Java?LockSupport的示例代碼

更新時間：2022年08月17日 11:44:40 作者：一無是處的研究僧

LockSupport給我們提供了一個非常強大的功能，它是線程阻塞最基本的元語，他可以將一個線程阻塞也可以將一個線程喚醒，因此經(jīng)常在并發(fā)的場景下進(jìn)行使用。本文將用60行代碼實現(xiàn)手寫LockSupport，需要的可以參考一下

前言

在JDK當(dāng)中給我們提供的各種并發(fā)工具當(dāng)中，比如ReentrantLock等等工具的內(nèi)部實現(xiàn)，經(jīng)常會使用到一個工具，這個工具就是LockSupport。LockSupport給我們提供了一個非常強大的功能，它是線程阻塞最基本的元語，他可以將一個線程阻塞也可以將一個線程喚醒，因此經(jīng)常在并發(fā)的場景下進(jìn)行使用。

LockSupport實現(xiàn)原理

在了解LockSupport實現(xiàn)原理之前我們先用一個案例來了解一下LockSupport的功能！

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
 
public class Demo {
 
  public static void main(String[] args) throws InterruptedException {
    Thread thread = new Thread(() -> {
      System.out.println("park 之前");
      LockSupport.park(); // park 函數(shù)可以將調(diào)用這個方法的線程掛起
      System.out.println("park 之后");
    });
    thread.start();
    TimeUnit.SECONDS.sleep(5);
    System.out.println("主線程休息了 5s");
    System.out.println("主線程 unpark thread");
    LockSupport.unpark(thread); // 主線程將線程 thread 喚醒 喚醒之后線程 thread 才可以繼續(xù)執(zhí)行
  }
}

上面的代碼的輸出如下：

park 之前
主線程休息了 5s
主線程 unpark thread
park 之后

乍一看上面的LockSupport的park和unpark實現(xiàn)的功能和await和signal實現(xiàn)的功能好像是一樣的，但是其實不然，我們來看下面的代碼：

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
 
public class Demo02 {
  public static void main(String[] args) throws InterruptedException {
    Thread thread = new Thread(() -> {
      try {
        TimeUnit.SECONDS.sleep(5);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      System.out.println("park 之前");
      LockSupport.park(); // 線程 thread 后進(jìn)行 park 操作 
      System.out.println("park 之后");
    });
    thread.start();
    System.out.println("主線程 unpark thread");
    LockSupport.unpark(thread); // 先進(jìn)行 unpark 操作
 
  }
}

上面代碼輸出結(jié)果如下：

主線程 unpark thread
park 之前
park 之后

在上面的代碼當(dāng)中主線程會先進(jìn)行unpark操作，然后線程thread才進(jìn)行park操作，這種情況下程序也可以正常執(zhí)行。但是如果是signal的調(diào)用在await調(diào)用之前的話，程序則不會執(zhí)行完成，比如下面的代碼：

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
 
public class Demo03 {
 
  private static final ReentrantLock lock = new ReentrantLock();
  private static final Condition condition = lock.newCondition();
 
  public static void thread() throws InterruptedException {
    lock.lock();
 
    try {
      TimeUnit.SECONDS.sleep(5);
      condition.await();
      System.out.println("等待完成");
    }finally {
      lock.unlock();
    }
  }
 
  public static void mainThread() {
    lock.lock();
    try {
      System.out.println("發(fā)送信號");
      condition.signal();
    }finally {
      lock.unlock();
      System.out.println("主線程解鎖完成");
    }
  }
 
  public static void main(String[] args) {
    Thread thread = new Thread(() -> {
      try {
        thread();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    });
    thread.start();
 
    mainThread();
  }
}

上面的代碼輸出如下：

發(fā)送信號
主線程解鎖完成

在上面的代碼當(dāng)中“等待完成“始終是不會被打印出來的，這是因為signal函數(shù)的調(diào)用在await之前，signal函數(shù)只會對在它之前執(zhí)行的await函數(shù)有效果，對在其后面調(diào)用的await是不會產(chǎn)生影響的。

那是什么原因?qū)е碌倪@個效果呢？

其實JVM在實現(xiàn)LockSupport的時候，內(nèi)部會給每一個線程維護(hù)一個計數(shù)器變量_counter，這個變量是表示的含義是“許可證的數(shù)量”，只有當(dāng)有許可證的時候線程才可以執(zhí)行，同時許可證最大的數(shù)量只能為1。當(dāng)調(diào)用一次park的時候許可證的數(shù)量會減一。當(dāng)調(diào)用一次unpark的時候計數(shù)器就會加一，但是計數(shù)器的值不能超過1。

當(dāng)一個線程調(diào)用park之后，他就需要等待一個許可證，只有拿到許可證之后這個線程才能夠繼續(xù)執(zhí)行，或者在park之前已經(jīng)獲得一個了一個許可證，那么它就不需要阻塞，直接可以執(zhí)行。

自己動手實現(xiàn)自己的LockSupport

實現(xiàn)原理

在前文當(dāng)中我們已經(jīng)介紹了locksupport的原理，它主要的內(nèi)部實現(xiàn)就是通過許可證實現(xiàn)的：

每一個線程能夠獲取的許可證的最大數(shù)目就是1。
當(dāng)調(diào)用unpark方法時，線程可以獲取一個許可證，許可證數(shù)量的上限是1，如果已經(jīng)有一個許可證了，那么許可證就不能累加。
當(dāng)調(diào)用park方法的時候，如果調(diào)用park方法的線程沒有許可證的話，則需要將這個線程掛起，直到有其他線程調(diào)用unpark方法，給這個線程發(fā)放一個許可證，線程才能夠繼續(xù)執(zhí)行。但是如果線程已經(jīng)有了一個許可證，那么線程將不會阻塞可以直接執(zhí)行。

自己實現(xiàn)LockSupport協(xié)議規(guī)定

在我們自己實現(xiàn)的Parker當(dāng)中我們也可以給每個線程一個計數(shù)器，記錄線程的許可證的數(shù)目，當(dāng)許可證的數(shù)目大于等于0的時候，線程可以執(zhí)行，反之線程需要被阻塞，協(xié)議具體規(guī)則如下：

初始線程的許可證的數(shù)目為0。
如果我們在調(diào)用park的時候，計數(shù)器的值等于1，計數(shù)器的值變?yōu)?，則線程可以繼續(xù)執(zhí)行。
如果我們在調(diào)用park的時候，計數(shù)器的值等于0，則線程不可以繼續(xù)執(zhí)行，需要將線程掛起，且將計數(shù)器的值設(shè)置為-1。
如果我們在調(diào)用unpark的時候，被unpark的線程的計數(shù)器的值等于0，則需要將計數(shù)器的值變?yōu)?。
如果我們在調(diào)用unpark的時候，被unpark的線程的計數(shù)器的值等于1，則不需要改變計數(shù)器的值，因為計數(shù)器的最大值就是1。
我們在調(diào)用unpark的時候，如果計數(shù)器的值等于-1，說明線程已經(jīng)被掛起了，則需要將線程喚醒，同時需要將計數(shù)器的值設(shè)置為0。

工具

因為涉及線程的阻塞和喚醒，我們可以使用可重入鎖ReentrantLock和條件變量Condition，因此需要熟悉這兩個工具的使用。

ReentrantLock 主要用于加鎖和開鎖，用于保護(hù)臨界區(qū)。

Condition.awat 方法用于將線程阻塞。

Condition.signal 方法用于將線程喚醒。

因為我們在unpark方法當(dāng)中需要傳入具體的線程，將這個線程發(fā)放許可證，同時喚醒這個線程，因為是需要針對特定的線程進(jìn)行喚醒，而condition喚醒的線程是不確定的，因此我們需要為每一個線程維護(hù)一個計數(shù)器和條件變量，這樣每個條件變量只與一個線程相關(guān)，喚醒的肯定就是一個特定的線程。我們可以使用HashMap進(jìn)行實現(xiàn)，鍵為線程，值為計數(shù)器或者條件變量。

具體實現(xiàn)

因此綜合上面的分析我們的類變量如下：

private final ReentrantLock lock; // 用于保護(hù)臨界去
private final HashMap<Thread, Integer> permits; // 許可證的數(shù)量
private final HashMap<Thread, Condition> conditions; // 用于喚醒和阻塞線程的條件變量

構(gòu)造函數(shù)主要對變量進(jìn)行賦值：

public Parker() {
  lock = new ReentrantLock();
  permits = new HashMap<>();
  conditions = new HashMap<>();
}

park方法

public void park() {
  Thread t = Thread.currentThread(); // 首先得到當(dāng)前正在執(zhí)行的線程
  if (conditions.get(t) == null) { // 如果還沒有線程對應(yīng)的condition的話就進(jìn)行創(chuàng)建
    conditions.put(t, lock.newCondition());
  }
  lock.lock();
  try {
    // 如果許可證變量還沒有創(chuàng)建 或者許可證等于0 說明沒有許可證了 線程需要被掛起
    if (permits.get(t) == null || permits.get(t) == 0) {
      permits.put(t, -1); // 同時許可證的數(shù)目應(yīng)該設(shè)置為-1
      conditions.get(t).await();
    }else if (permits.get(t) > 0) {
      permits.put(t, 0); // 如果許可證的數(shù)目大于0 也就是為1 說明線程已經(jīng)有了許可證因此可以直接被放行 但是需要消耗一個許可證
    }
  } catch (InterruptedException e) {
    e.printStackTrace();
  } finally {
    lock.unlock();
  }
}

unpark方法

public void unpark(Thread thread) {
  Thread t = thread; // 給線程 thread 發(fā)放一個許可證
  lock.lock();
  try {
    if (permits.get(t) == null) // 如果還沒有創(chuàng)建許可證變量 說明線程當(dāng)前的許可證數(shù)量等于初始數(shù)量也就是0 因此方法許可證之后 許可證的數(shù)量為 1
      permits.put(t, 1);
    else if (permits.get(t) == -1) { // 如果許可證數(shù)量為-1，則說明肯定線程 thread 調(diào)用了park方法，而且線程 thread已經(jīng)被掛起了 因此在 unpark 函數(shù)當(dāng)中不急需要將許可證數(shù)量這是為0 同時還需要將線程喚醒
      permits.put(t, 0);
      conditions.get(t).signal();
    }else if (permits.get(t) == 0) { // 如果許可證數(shù)量為0 說明線程正在執(zhí)行 因此許可證數(shù)量加一
      permits.put(t, 1);
    } // 除此之外就是許可證為1的情況了 在這種情況下是不需要進(jìn)行操作的 因為許可證最大的數(shù)量就是1
  }finally {
    lock.unlock();
  }
}

完整代碼

import java.util.HashMap;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
 
public class Parker {
 
  private final ReentrantLock lock;
  private final HashMap<Thread, Integer> permits;
  private final HashMap<Thread, Condition> conditions;
 
  public Parker() {
    lock = new ReentrantLock();
    permits = new HashMap<>();
    conditions = new HashMap<>();
  }
 
  public void park() {
    Thread t = Thread.currentThread();
    if (conditions.get(t) == null) {
      conditions.put(t, lock.newCondition());
    }
    lock.lock();
    try {
      if (permits.get(t) == null || permits.get(t) == 0) {
        permits.put(t, -1);
        conditions.get(t).await();
      }else if (permits.get(t) > 0) {
        permits.put(t, 0);
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    } finally {
      lock.unlock();
    }
  }
 
  public void unpark(Thread thread) {
    Thread t = thread;
    lock.lock();
    try {
      if (permits.get(t) == null)
        permits.put(t, 1);
      else if (permits.get(t) == -1) {
        permits.put(t, 0);
        conditions.get(t).signal();
      }else if (permits.get(t) == 0) {
        permits.put(t, 1);
      }
    }finally {
      lock.unlock();
    }
  }
}

JVM實現(xiàn)一瞥

其實在JVM底層對于park和unpark的實現(xiàn)也是基于鎖和條件變量的，只不過是用更加底層的操作系統(tǒng)和libc(linux操作系統(tǒng))提供的API進(jìn)行實現(xiàn)的。雖然API不一樣，但是原理是相仿的，思想也相似。

比如下面的就是JVM實現(xiàn)的unpark方法：

void Parker::unpark() {
  int s, status;
  // 進(jìn)行加鎖操作 相當(dāng)于 可重入鎖的 lock.lock()
  status = pthread_mutex_lock(_mutex);
  assert (status == 0, "invariant");
  s = _counter;
  _counter = 1;
  if (s < 1) {
    // 如果許可證小于 1 進(jìn)行下面的操作
    if (WorkAroundNPTLTimedWaitHang) {
      // 這行代碼相當(dāng)于 condition.signal() 喚醒線程
      status = pthread_cond_signal (_cond);
      assert (status == 0, "invariant");
      // 解鎖操作 相當(dāng)于可重入鎖的 lock.unlock()
      status = pthread_mutex_unlock(_mutex);
      assert (status == 0, "invariant");
    } else {
      status = pthread_mutex_unlock(_mutex);
      assert (status == 0, "invariant");
      status = pthread_cond_signal (_cond);
      assert (status == 0, "invariant");
    }
  } else {
    // 如果有許可證 也就是 s == 1 那么不許要將線程掛起
    // 解鎖操作 相當(dāng)于可重入鎖的 lock.unlock()
    pthread_mutex_unlock(_mutex);
    assert (status == 0, "invariant");
  }
}

JVM實現(xiàn)的park方法，如果沒有許可證也是會將線程掛起的：

總結(jié)

在本篇文章當(dāng)中主要介紹啦lock support的用法以及它的大致原理，以及介紹啦我們自己該如何實現(xiàn)類似lock support的功能，并且定義了我們自己實現(xiàn)lock support的大致協(xié)議，整個過程還是比較清晰的，我們只是實現(xiàn)了lock support當(dāng)中兩個核心方法，其他的方法其實也類似，原理差不多，在這里咱就實現(xiàn)一個乞丐版的lock support的吧?。?！

使用鎖和條件變量進(jìn)行線程的阻塞和喚醒。

使用Thread.currentThread()方法得到當(dāng)前正在執(zhí)行的線程。

使用HashMap去存儲線程和許可證以及條件變量的關(guān)系。

以上就是手寫Java LockSupport的示例代碼的詳細(xì)內(nèi)容，更多關(guān)于Java LockSupport的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: