Implementierungsmethode für handgeschriebenes Java LockSupport

Vorwort

Unter den verschiedenen Parallelitätstools, die uns im JDK zur Verfügung gestellt werden, wie z. B. der internen Implementierung von ReentrantLock und anderen Tools, wird häufig ein Tool verwendet, und dieses Tool ist LockSupport. LockSupport bietet uns eine sehr leistungsstarke Funktion. Es ist das grundlegendste Grundelement für die Thread-Blockierung. Es kann einen Thread blockieren oder aufwecken und wird daher häufig in gleichzeitigen Szenarien verwendet.

LockSupport-Implementierungsprinzip

Bevor wir das LockSupport-Implementierungsprinzip verstehen, verwenden wir zunächst einen Fall, um die Funktionen von LockSupport zu verstehen!

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 函数可以将调用这个方法的线程挂起
      System.out.println("park 之后");
    });
    thread.start();
    TimeUnit.SECONDS.sleep(5);
    System.out.println("主线程休息了 5s");
    System.out.println("主线程 unpark thread");
    LockSupport.unpark(thread); // 主线程将线程 thread 唤醒 唤醒之后线程 thread 才可以继续执行
  }
}

Die Ausgabe des obigen Codes lautet wie folgt:

park Vorher
der Hauptthread ruhte 5 Sekunden lang
Hauptthread entparken Thread
parken nach

Auf den ersten Blick warten die oben genannten Park- und Unpark-Implementierungsfunktionen von LockSupport und Signalimplementierung Die Funktionen scheinen gleich zu sein, sind es aber tatsächlich nicht. Schauen wir uns den folgenden Code an:

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 后进行 park 操作 
      System.out.println("park 之后");
    });
    thread.start();
    System.out.println("主线程 unpark thread");
    LockSupport.unpark(thread); // 先进行 unpark 操作
 
  }
}

Die Ausgabe des obigen Codes lautet wie folgt:

Hauptthread entparken Thread
vorher parken
park nach

Im obigen Code wird zuerst der Entparkvorgang ausgeführt, und dann führt der Thread den Parkvorgang aus. In diesem Fall kann das Programm auch normal ausgeführt werden. Wenn der Signalaufruf jedoch vor dem Warteaufruf erfolgt, wird das Programm beispielsweise nicht ausgeführt:

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("发送信号");
      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();
  }
}

Die Ausgabe des obigen Codes lautet wie folgt:

Signal senden
Die Entsperrung des Hauptthreads ist abgeschlossen

Im obigen Code wird „Warten auf Abschluss“ nie ausgedruckt. Dies liegt daran, dass die Signalfunktion vor dem Aufruf aufgerufen wird. Die Signalfunktion hat nur Auswirkungen auf die zuvor ausgeführte Wartefunktion, nicht jedoch auf das Warten Die danach aufgerufene Funktion hat Auswirkungen.

Was verursacht diesen Effekt?

Tatsächlich verwaltet die JVM intern eine Zählervariable _counter für jeden Thread. Diese Variable stellt die „Anzahl der Lizenzen“ dar. Der Thread kann nur gleichzeitig ausgeführt werden Die maximale Anzahl an Lizenzen kann nur 1 betragen. Bei einmaligem Parkaufruf wird die Anzahl der Lizenzen um eine reduziert. Wenn unpark einmal aufgerufen wird, wird der Zähler um eins erhöht, der Wert des Zählers darf jedoch 1 nicht überschreiten.

Wenn ein Thread Park aufruft, muss er auf eine Lizenz warten. Erst nachdem er die Lizenz erhalten hat, kann der Thread weiter ausgeführt werden. Wenn er vor dem Parken eine Lizenz erhalten hat, muss er nicht blockiert werden und kann direkt ausgeführt werden .

Implementieren Sie Ihren eigenen LockSupport

Implementierungsprinzip

Im vorherigen Artikel haben wir das Prinzip von LockSupport vorgestellt. Seine interne Hauptimplementierung wird durch Lizenzen realisiert:

• Die Lizenz, die jeder Thread erhalten kann Die maximale Anzahl ist 1.

• Wenn die Unpark-Methode aufgerufen wird, kann der Thread eine Lizenz erhalten. Die Obergrenze der Anzahl der Lizenzen beträgt 1. Wenn bereits eine Lizenz vorhanden ist, können die Lizenzen nicht akkumuliert werden.

• Wenn der Thread, der die Park-Methode aufruft, keine Lizenz hat, muss der Thread angehalten werden, bis andere Threads die Unpark-Methode aufrufen und dem Thread eine Lizenz erteilen, bevor der Thread fortfahren kann . Wenn der Thread jedoch bereits über eine Lizenz verfügt, wird der Thread nicht blockiert und kann direkt ausgeführt werden.

Implementieren Sie die LockSupport-Protokollbestimmungen selbst.

In unserer eigenen Implementierung von Parker können wir jedem Thread auch einen Zähler zuweisen, um die Anzahl der Lizenzen für den Thread aufzuzeichnen, wenn die Anzahl der Lizenzen größer oder gleich 0 ist , der Thread kann ausgeführt werden, andernfalls muss der Thread blockiert werden. Die spezifischen Regeln des Protokolls lauten wie folgt:

• Die Anzahl der Lizenzen für den ersten Thread beträgt 0.

• Wenn beim Parkaufruf der Zählerwert gleich 1 ist und der Zählerwert 0 wird, kann der Thread weiter ausgeführt werden.

• Wenn der Zählerwert beim Aufrufen von Park gleich 0 ist, kann der Thread nicht weiter ausgeführt werden, der Thread muss angehalten werden und der Zählerwert wird auf -1 gesetzt.

• Wenn beim Aufruf von unpark der Zählerwert des nicht geparkten Threads gleich 0 ist, muss der Zählerwert auf 1 geändert werden.

• Wenn der Zählerwert des nicht geparkten Threads beim Aufruf von unpark gleich 1 ist, besteht keine Notwendigkeit, den Zählerwert zu ändern, da der Maximalwert des Zählers 1 ist.

• Wenn wir Unpark aufrufen und der Zählerwert gleich -1 ist, bedeutet dies, dass der Thread angehalten wurde und der Thread aktiviert werden muss und der Zählerwert auf 0 gesetzt werden muss.

Tools

Da es sich um das Blockieren und Aufwecken von Threads handelt, können wir die Wiedereintrittssperre ReentrantLock und die Bedingungsvariablen Condition verwenden. Daher müssen wir mit der Verwendung dieser beiden Tools vertraut sein.

ReentrantLock wird hauptsächlich zum Sperren und Entsperren sowie zum Schutz kritischer Bereiche verwendet.

Condition.awat-Methode wird verwendet, um den Thread zu blockieren.

Die Condition.signal-Methode wird verwendet, um den Thread aufzuwecken.

因为我们在unpark方法当中需要传入具体的线程，将这个线程发放许可证，同时唤醒这个线程，因为是需要针对特定的线程进行唤醒，而condition唤醒的线程是不确定的，因此我们需要为每一个线程维护一个计数器和条件变量，这样每个条件变量只与一个线程相关，唤醒的肯定就是一个特定的线程。我们可以使用HashMap进行实现，键为线程，值为计数器或者条件变量。

具体实现

因此综合上面的分析我们的类变量如下：

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<>();
}

park方法

public void park() {
  Thread t = Thread.currentThread(); // 首先得到当前正在执行的线程
  if (conditions.get(t) == null) { // 如果还没有线程对应的condition的话就进行创建
    conditions.put(t, lock.newCondition());
  }
  lock.lock();
  try {
    // 如果许可证变量还没有创建 或者许可证等于0 说明没有许可证了 线程需要被挂起
    if (permits.get(t) == null || permits.get(t) == 0) {
      permits.put(t, -1); // 同时许可证的数目应该设置为-1
      conditions.get(t).await();
    }else if (permits.get(t) > 0) {
      permits.put(t, 0); // 如果许可证的数目大于0 也就是为1 说明线程已经有了许可证因此可以直接被放行 但是需要消耗一个许可证
    }
  } catch (InterruptedException e) {
    e.printStackTrace();
  } finally {
    lock.unlock();
  }
}

unpark方法

public void unpark(Thread thread) {
  Thread t = thread; // 给线程 thread 发放一个许可证
  lock.lock();
  try {
    if (permits.get(t) == null) // 如果还没有创建许可证变量 说明线程当前的许可证数量等于初始数量也就是0 因此方法许可证之后 许可证的数量为 1
      permits.put(t, 1);
    else if (permits.get(t) == -1) { // 如果许可证数量为-1，则说明肯定线程 thread 调用了park方法，而且线程 thread已经被挂起了 因此在 unpark 函数当中不急需要将许可证数量这是为0 同时还需要将线程唤醒
      permits.put(t, 0);
      conditions.get(t).signal();
    }else if (permits.get(t) == 0) { // 如果许可证数量为0 说明线程正在执行 因此许可证数量加一
      permits.put(t, 1);
    } // 除此之外就是许可证为1的情况了 在这种情况下是不需要进行操作的 因为许可证最大的数量就是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实现一瞥

其实在JVM底层对于park和unpark的实现也是基于锁和条件变量的，只不过是用更加底层的操作系统和libc(linux操作系统)提供的API进行实现的。虽然API不一样，但是原理是相仿的，思想也相似。

比如下面的就是JVM实现的unpark方法：

void Parker::unpark() {
  int s, status;
  // 进行加锁操作 相当于 可重入锁的 lock.lock()
  status = pthread_mutex_lock(_mutex);
  assert (status == 0, "invariant");
  s = _counter;
  _counter = 1;
  if (s < 1) {
    // 如果许可证小于 1 进行下面的操作
    if (WorkAroundNPTLTimedWaitHang) {
      // 这行代码相当于 condition.signal() 唤醒线程
      status = pthread_cond_signal (_cond);
      assert (status == 0, "invariant");
      // 解锁操作 相当于可重入锁的 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 那么不许要将线程挂起
    // 解锁操作 相当于可重入锁的 lock.unlock()
    pthread_mutex_unlock(_mutex);
    assert (status == 0, "invariant");
  }
}

JVM实现的park方法，如果没有许可证也是会将线程挂起的：

Das obige ist der detaillierte Inhalt vonImplementierungsmethode für handgeschriebenes Java LockSupport. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!