spb 2020-05-10
前面一篇文章中说明了Object的阻塞唤醒机制,今天我们要讲解另一个类LockSupport,在AQS中你能看见它的身影,所以需要提前了解其实现和使用机制,便于后面深入AQS的学习
JDK版本号:1.8.0_171
在源码阅读之前希望大家先去阅读几遍注释,其中介绍了LockSupport的设计,实现和使用机制,这里进行简单说明下:
上述有些术语可能令人困惑,这里我们通俗点说,首先需要理解permit(许可),这里也就是相当于一个变量标志,有兴趣可查看Hotspot源码
HotSpot Parker用condition和mutex维护了一个_counter变量,park时,变量_counter置为0,unpark时,变量_counter置为1
连续两次调用park操作,变量不会变成2,还是1,也就是说的不能叠加,你可以自己写代码验证,因为维护的是一个变量标识更新,所以park和unpark的调用没有先后顺序限制:
简单示例代码如下:
Thread test = new Thread(new Runnable() { @Override public void run() { System.out.println("start"); LockSupport.park(this);// _counter为0,阻塞 System.out.println("end"); } }); test.start(); Thread.sleep(3000); System.out.println("ready notify"); // 线程对应的_counter置为1,同时唤醒阻塞的线程,唤醒的线程消耗掉1置为0 LockSupport.unpark(test);
在AbstractQueuedSynchronizer中使用了LockSupport实现线程阻塞和唤醒操作,所以有必要先进行了解,怎么通过LockSupport实现FIFO互斥锁呢?源码注释处已经提供了思路,非队首线程或者不能更新锁标识的都需要被阻塞,还是挺巧妙的,可以好好理解理解
public class FIFOMutex { public static void main(String[] args) { FIFOMutex lock = new FIFOMutex(); new Thread(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread()+"111"); lock.lock(); System.out.println(Thread.currentThread()+"111"); } }).start(); new Thread(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread()+"222"); lock.lock(); System.out.println(Thread.currentThread()+"222"); } }).start(); new Thread(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread()+"333"); lock.lock(); System.out.println(Thread.currentThread()+"333"); } }).start(); Thread.sleep(1000); lock.unlock(); Thread.sleep(1000); lock.unlock(); } private final AtomicBoolean locked = new AtomicBoolean(false); private final Queue<Thread> waiters = new ConcurrentLinkedQueue<Thread>(); public void lock() { boolean wasInterrupted = false; Thread current = Thread.currentThread(); waiters.add(current); // Block while not first in queue or cannot acquire lock // 非队首线程或者CAS获取不到锁标识则进行阻塞 while (waiters.peek() != current || !locked.compareAndSet(false, true)) { LockSupport.park(this); if (Thread.interrupted()) // ignore interrupts while waiting wasInterrupted = true; } waiters.remove(); if (wasInterrupted) // reassert interrupt status on exit current.interrupt(); } public void unlock() { locked.set(false); LockSupport.unpark(waiters.peek()); } }
常量部分通过CAS来完成操作,没什么需要多说的,简单理解就好,不是重点
// Hotspot implementation via intrinsics API private static final sun.misc.Unsafe UNSAFE; private static final long parkBlockerOffset; private static final long SEED; private static final long PROBE; private static final long SECONDARY; static { try { UNSAFE = sun.misc.Unsafe.getUnsafe(); Class<?> tk = Thread.class; parkBlockerOffset = UNSAFE.objectFieldOffset (tk.getDeclaredField("parkBlocker")); SEED = UNSAFE.objectFieldOffset (tk.getDeclaredField("threadLocalRandomSeed")); PROBE = UNSAFE.objectFieldOffset (tk.getDeclaredField("threadLocalRandomProbe")); SECONDARY = UNSAFE.objectFieldOffset (tk.getDeclaredField("threadLocalRandomSecondarySeed")); } catch (Exception ex) { throw new Error(ex); } }
空的私有构造方法,不能被外部实例化
private LockSupport() {} // Cannot be instantiated.
大量调用了UNSAFE的native方法,有兴趣的可以去找HotSpot源码来深入学习,我们这里仅做了解使用即可
park相关方法中被调用,记录阻塞的对象,也就是监视和阻断工具查原因时保存的对象
private static void setBlocker(Thread t, Object arg) { // Even though volatile, hotspot doesn‘t need a write barrier here. UNSAFE.putObject(t, parkBlockerOffset, arg); }
简单理解为唤醒对应的thread线程是不正确的,实际上,即使thread线程未调用park操作阻塞这里unpark操作也是可以进行的,使得thread线程的permit处于可用状态,那么之后thread线程调用park线程将不会被阻塞,因为permit可用,参考前言写些代码多理解理解
public static void unpark(Thread thread) { if (thread != null) UNSAFE.unpark(thread); }
在permit处于不可用状态时,阻塞当前线程,同时可传入blocker信息,同时注意被唤醒条件有以下三种:
被唤醒的原因不会被返回,所以需要调用方自行检查是什么原因
public static void park() { UNSAFE.park(false, 0L); } public static void park(Object blocker) { Thread t = Thread.currentThread(); // 这个地方在阻塞前保存了blocker信息 setBlocker(t, blocker); UNSAFE.park(false, 0L); // 被唤醒之后被将blocker信息置空 setBlocker(t, null); }
在permit处于不可用状态时,阻塞当前线程nanos毫秒,同时可传入blocker信息,唤醒机制和park()类似,除了多了一个超时条件,当然这里是超时自动唤醒的机制
public static void parkNanos(long nanos) { if (nanos > 0) UNSAFE.park(false, nanos); } public static void parkNanos(Object blocker, long nanos) { if (nanos > 0) { Thread t = Thread.currentThread(); setBlocker(t, blocker); UNSAFE.park(false, nanos); setBlocker(t, null); } }
在permit处于不可用状态时,阻塞当前线程到deadline时间点,同时可传入blocker信息,与parkNanos类似
public static void parkUntil(long deadline) { UNSAFE.park(true, deadline); } public static void parkUntil(Object blocker, long deadline) { Thread t = Thread.currentThread(); setBlocker(t, blocker); UNSAFE.park(true, deadline); setBlocker(t, null); }
获取线程t的blocker对象信息,也就是被阻塞前通过setBlocker(t, blocker)传入的对象信息
public static Object getBlocker(Thread t) { if (t == null) throw new NullPointerException(); return UNSAFE.getObjectVolatile(t, parkBlockerOffset); }
这个方法是由于多线程随机数生成器ThreadLocalRandom的package访问权限限制不能被这个包下的类使用,复制了一份实现出来,在StampedLock中被使用,有兴趣可以去了解,以后会在StampedLock的源码中进行说明
/** * Returns the pseudo-randomly initialized or updated secondary seed. * Copied from ThreadLocalRandom due to package access restrictions. */ static final int nextSecondarySeed() { int r; Thread t = Thread.currentThread(); if ((r = UNSAFE.getInt(t, SECONDARY)) != 0) { r ^= r << 13; // xorshift r ^= r >>> 17; r ^= r << 5; } else if ((r = java.util.concurrent.ThreadLocalRandom.current().nextInt()) == 0) r = 1; // avoid zero UNSAFE.putInt(t, SECONDARY, r); return r; }
那么LockSupport的阻塞唤醒机制和Object的阻塞唤醒机制有什么区别呢?
本文分析了LockSupport的使用和源码,简单说明了Hotspot源码中对应的实现机制,方便各位理解,本质上而言还是很好理解的,其实对于我们而言更重要的在于使用,在线程阻塞唤醒机制上的使用需要大家多理解理解,下篇文章我们就开始进行AQS的源码学习了,当然要好好理解下LockSupport
以上内容如有问题欢迎指出,笔者验证后将及时修正,谢谢