目录
blocker对象分析
public class LockSupport {
public static void park(Object blocker) {
// 1. 设置blocker
Thread t = Thread.currentThread();
setBlocker(t, blocker);
// 2. 执行park操作
U.park(false, 0L);
// 3. 清除blocker
setBlocker(t, null);
}
}再park方法里会执行setBlocker方法，再执行park方法，那这个setBlocker方法到底有什么用呢？
private static void setBlocker(Thread t, Object arg) {
U.putReferenceOpaque(t, PARKBLOCKER, arg);
}基本作用
这个方法的主要作用是设置线程对象的parkBlocker字段，用于记录线程被谁阻塞。
在park方法需要传入一个对象，一般都会传入this对象，在park方法里会将这个this对象设置到Thread类的parkBlocker对象里，可用来监控和诊断，可以通过线程对象的parkBlocker对象判断该线程是由什么造成的一个阻塞。
内存写入：将一个对象引用（arg）写入到线程对象（t）的特定内存偏移位（PARKBLOCKER），这是一个直接操作内存的操作，绕过了Java的访问控制
Opaque语义："Opaque"表示这是一个有序的写入操作，不会与其他内存操作重排序，保证写入的可见性
// 伪代码展示内存屏障效果
void putReferenceOpaque(Object o, long offset, Object value) {
// StoreStore屏障
// 确保之前的写操作都已完成
storeStoreBarrier();
// 写入内存
putReference(o, offset, value);
// StoreLoad屏障
// 确保写入对其他线程可见
storeLoadBarrier();
}内存屏障效果
// 伪代码展示内存屏障效果
void putReferenceOpaque(Object o, long offset, Object value) {
// StoreStore屏障
// 确保之前的写操作都已完成
storeStoreBarrier();
// 写入内存
putReference(o, offset, value);
// StoreLoad屏障
// 确保写入对其他线程可见
storeLoadBarrier();
}可见性保证：写入操作对其他线程立即可见，不会被CPU缓存影响，直接写入主内存
顺序性保证：写入操作不会与之前的写操作重排序，写入操作不会与之后的读操作重排序，维护程序的顺序性语义
class Thread {
// ... 其他字段
// parkBlocker字段，用于存储导致线程阻塞的对象
private volatile Object parkBlocker;
// ... 其他字段
}parkBlocker对象会保存在线程里面，看看下面我写的引用代码实现。
写入过程
// Unsafe的操作过程
public class Unsafe {
// PARKBLOCKER是parkBlocker字段在Thread类中的内存偏移量
private static final long PARKBLOCKER = ...;
public void putReferenceOpaque(Object o, long offset, Object value) {
// 1. o是Thread对象的内存起始地址
// 2. offset是parkBlocker字段相对于对象起始位置的偏移量
// 3. value是要存储的对象引用
// 直接将value写入到 o + offset 的内存位置
}
}实际应用
// AQS中的使用示例
public abstract class AbstractQueuedSynchronizer {
public final void acquire(int arg) {
if (!tryAcquire(arg)) {
// 将当前线程封装成节点
Node node = addWaiter(Node.EXCLUSIVE);
// 进入等待队列
acquireQueued(node, arg);
}
}
final boolean acquireQueued(final Node node, int arg) {
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt()) {
interrupted = true;
}
}
} catch (RuntimeException ex) {
cancelAcquire(node);
throw ex;
}
}
private final boolean parkAndCheckInterrupt() {
// 使用当前的AQS对象作为blocker
LockSupport.park(this);
return Thread.interrupted();
}
}其他线程可见性
// 其他线程可以立即看到写入的值
Thread t1 = new Thread(() -> {
Object blocker = new Object();
LockSupport.park(blocker);
});
Thread t2 = new Thread(() -> {
// 可以立即看到t1的blocker对象
Object b = LockSupport.getBlocker(t1);
});诊断和监控
public class ThreadMonitor {
public void monitorThreads() {
Thread[] threads = getAllThreads();
for (Thread t : threads) {
Object blocker = LockSupport.getBlocker(t);
if (blocker != null) {
if (blocker instanceof AbstractQueuedSynchronizer) {
// 线程在等待AQS锁
AbstractQueuedSynchronizer aqs = (AbstractQueuedSynchronizer)blocker;
System.out.println("线程 " + t.getName() +
" 正在等待锁 " + aqs);
} else if (blocker instanceof ReentrantLock) {
// 线程在等待ReentrantLock
System.out.println("线程 " + t.getName() +
" 正在等待ReentrantLock");
} else if (blocker instanceof Condition) {
// 线程在等待Condition
System.out.println("线程 " + t.getName() +
" 正在等待Condition");
}
}
}
}
}