CLH锁的机制与实现

目标

了解CLH的机制并简单实现，为学习AbstractQueuedSynchronizer打下基础。

我们大致了解AQS是jdk实现各种内置同步器（ReentrantLock、CountDownLatch、Semaphore、CyclicBarrier）的基础，也是我们想要自定义同步器需要借助的工具，也知道AQS的内部实现机制是CLH锁。

现在我们就先了解了解CLH的机制和简单实现。

CLH lock is Craig, Landin, and Hagersten (CLH) locks, CLH lock is a spin lock, can ensure no hunger, provide fairness first come first service.

CLH是人名简写。这种锁本质是自旋锁，确保无饥饿、公平地先进先出竞争锁。

设计思路

CLH锁的数据结构：

下面说的代理，非代理模式，是代表的意思。

内部类Node：Node节点代理一个竞争者，它有一个标记自己是否处于竞争或锁定状态的标志，true：正在竞争或已经获得锁；false：已经释放锁
成员变量：ThreadLocal<Node> current;——每个线程调用current.get()将获得自己参与竞争的代理，即一个Node实例
成员变量：ThreadLocal<Node> pre;——每个线程调用pre.get()将获得上一个竞争者
成员变量：AtomicReference<Node> tail;——维持在锁内部的最后一个竞争者节点

CLH锁的行为——实现Lock接口

思路

1、一个全局的CLHLock

2、每个线程都可以调用其lock方法，但只有一个线程能获得锁，其余线程等待

3、控制的方式是：

3.1 每次lock，先找到当前线程的代理节点（没有就新建），原子性地将其设置为tail
3.2 preNode指向原tail
3.3 循环判定preNode的锁定状态，如果已经解锁，while循环终止
3.4 每次unlock只需将锁定状态设置为false

代码与测试

package org.lanqiao.concurrent.syn_zer;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class CLHLock implements Lock {
/**Node可以认为是线程参与竞争的代理人，每个竞争的线程在Lock中都有一个作为代表的node对象
* 获取锁的前提是node的上一个node解锁了（locked==false）*/
private static final class Node {
boolean locked;
}

// 当前线程第一次get时，就会实例化一个Node并且放入线程本地变量
private ThreadLocal<Node> current = new ThreadLocal() {
@Override
protected Object initialValue() {
return new Node();
}
};
//pre用于记录排队节点的前驱
private ThreadLocal<Node> pre = new ThreadLocal<>();
//这记录了Node队列最后一个排队的Node，初始化时是一个unlock的node
private AtomicReference<Node> tail = new AtomicReference<>( new Node() );

/**获得锁*/
@Override
public void lock() {
// 当前线程的代理排队节点
Node myNode = current.get();
// 设置锁定状态
myNode.locked = true;
// 获得排队队列的最末一个节点，同时将tail指针指向当前线程的node
//这里获得并重新设置，是原子的，多个线程同时试图将自己的节点加入末尾，只有一个线程能成功
Node lastNode = tail.getAndSet( myNode );
//记录此前最末的节点到线程本地变量中
pre.set( lastNode );
//前一个node处于lock状态，当前线程死循环自旋
while (lastNode.locked) {
}
}

@Override
public void lockInterruptibly() throws InterruptedException {

}

@Override
public boolean tryLock() {
// 当前线程的节点
Node myNode = current.get();
// 设置锁定状态
myNode.locked = false;
// 获得排队队列的最末一个节点，同时将tail指针指向当前线程的node
//这里获得并重新设置，是原子的，多个线程同时视图将自己的节点加入末尾，只有一个线程能成功
Node lastNode = tail.getAndSet( myNode );
//记录此前最末的节点到线程本地变量中
pre.set( lastNode );
//前一个node处于lock状态，当前线程死循环自旋
return !(lastNode.locked);
}

@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return false;
}

@Override
public void unlock() {
// 当前线程的节点
Node myNode = current.get();
// 设置锁定状态为false，后继节点（线程）获得锁
myNode.locked = false;

pre.set( null );
}

@Override
public Condition newCondition() {
return null;
}

public static void main(String[] args) {
ExecutorService service = Executors.newFixedThreadPool( 10 );
CLHLock lock = new CLHLock();
for (int i = 0; i < 10; i++) {
service.submit( () -> {
lock.lock();
System.out.println( Thread.currentThread().getName() + ":I got the lock,but I do not release it." );
//  程序永远得不到退出
//lock.unlock();
//  不仅如此，因为所有的线程都在自旋中，cpu一致处于繁忙状态，会导致假死
} );
}
service.shutdown();
}
}

当然AQS中Node的实现要比这负责得多，我们只是通过简单的代码来了解其思想。