ホームページ  >  記事  >  Java  >  手書きJava LockSupportの実装方法

手書きJava LockSupportの実装方法

WBOY
WBOY転載
2023-05-07 08:25:06548ブラウズ

    まえがき

    ReentrantLock の内部実装やその他のツールなど、JDK によって提供されるさまざまな同時実行ツールの中で、よく使用されるツールがあります。 、このツールは LockSupport です。 LockSupport は非常に強力な機能を提供します。これはスレッド ブロックの最も基本的なプリミティブです。スレッドをブロックしたり、スレッドをウェイクアップしたりできるため、同時実行シナリオでよく使用されます。

    LockSupport の実装原理

    LockSupport の実装原理を理解する前に、まずケースを使用して 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 函数可以将调用这个方法的线程挂起
          System.out.println("park 之后");
        });
        thread.start();
        TimeUnit.SECONDS.sleep(5);
        System.out.println("主线程休息了 5s");
        System.out.println("主线程 unpark thread");
        LockSupport.unpark(thread); // 主线程将线程 thread 唤醒 唤醒之后线程 thread 才可以继续执行
      }
    }

    上記のコードの出力は次のとおりです。

    Before park
    The main threadrested for 5s
    The main thread unpark thread
    After park

    LockSupport の park と unpark で実装される関数と、await と signal で実装される関数は一見同じように見えますが、実際は同じではありません。次のコードを見てみましょう。

    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 操作
     
      }
    }

    上記のコード出力 結果は次のとおりです。

    Main thread unpark thread
    park before
    park after

    In上記のコードでは、メインスレッドが最初にパーク解除操作を実行し、次にスレッドスレッドがパーク操作のみを実行します。この場合、プログラムも正常に実行できます。ただし、シグナル呼び出しが await 呼び出しの前にある場合、プログラムは実行されません。たとえば、次のコード:

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

    上記のコードの出力は次のとおりです:

    #シグナル送信

    メインスレッドのロック解除が完了しました

    #上記のコードでは、「完了を待っています」というメッセージは出力されません。これは、await の前にシグナル関数が呼び出され、シグナル関数が呼び出されるからです。前に実行された await 関数は影響しますが、その後に呼び出される await には影響しません。

    この効果の理由は何ですか?

    実際には、JVM が LockSupport を実装すると、スレッドごとにカウンタ変数 _counter が内部的に維持されます。この変数は「ライセンス数」を表します。ライセンスがある場合にのみ、スレッドはただし、同時に実行できるライセンスの最大数は 1 つだけです。パークを 1 回呼び出すと、ライセンスの数が 1 つ減ります。 unpark が 1 回呼び出されると、カウンターは 1 つ増加しますが、カウンターの値は 1 を超えることはできません。

    スレッドがパークを呼び出すとき、スレッドはライセンスを待つ必要があります。ライセンスを取得した後にのみ、スレッドは実行を続行できます。または、パーク前にライセンスが取得されている場合は、実行を継続できません。ブロックされていても、直接実行できます。

    自分で独自の LockSupport を実装する

    実装原理

    前の記事では、ロックサポートの原理を紹介しました。その主な内部実装はライセンスによって実現されます。

      各スレッドが取得できるライセンスの最大数は 1 です。
    • unpark メソッドが呼び出されると、スレッドはライセンスを取得できます。ライセンス数の上限は 1 です。すでにライセンスがある場合、ライセンスを蓄積することはできません。
    • park メソッドを呼び出すときに、park メソッドを呼び出すスレッドにライセンスがない場合、別のスレッドが unpark メソッドを呼び出してこのスレッドに発行するまで、このスレッドを一時停止する必要があります。スレッドの実行を続行するにはライセンスが必要です。ただし、スレッドがすでにライセンスを持っている場合、スレッドはブロックされず、直接実行できます。
    • LockSupport プロトコル規制を独自に実装する

    独自の Parker 実装では、スレッドのライセンス数を記録するカウンターを各スレッドに与えることもできます。ライセンスの数が 0 以上の場合、スレッドは実行できますが、それ以外の場合はスレッドをブロックする必要があります。プロトコルの特定のルールは次のとおりです。 #初期スレッドのライセンス数は0です。

      カウンタ値が 1 に等しく、park を呼び出したときにカウンタ値が 0 になった場合、スレッドは実行を続行できます。
    • park を呼び出したときにカウンタ値が 0 の場合、スレッドは実行を続行できません。スレッドを一時停止する必要があり、カウンタ値は -1 に設定されます。
    • unpark を呼び出したときに、パーク解除されたスレッドのカウンタ値が 0 の場合、カウンタ値を 1 に変更する必要があります。
    • unpark を呼び出したときに、パーク解除されたスレッドのカウンターの値が 1 に等しい場合、カウンターの値を変更する必要はありません。カウンタは1です。
    • unpark を呼び出したときに、カウンター値が -1 に等しい場合、スレッドが一時停止されていることを意味するため、スレッドを起動してカウンター値を回復する必要があります。 0に設定します。
    • ツール
    • スレッドのブロックとウェイクアップが含まれるため、リエントラント ロック ReentrantLock と条件変数 Condition を使用できるため、これらの使用法に慣れる必要があります。 2つのツール。

    • ReentrantLock は主にロックとロック解除に使用され、重要なセクションを保護するために使用されます。

    Condition.awat メソッドはスレッドをブロックするために使用されます。

    Condition.signal メソッドは、スレッドを起動するために使用されます。

    因为我们在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方法,如果没有许可证也是会将线程挂起的:

    手書きJava LockSupportの実装方法

    以上が手書きJava LockSupportの実装方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

    声明:
    この記事はyisu.comで複製されています。侵害がある場合は、admin@php.cn までご連絡ください。