死磕 java同步系列之AQS起篇

static final class Node {    // 标识一个节点是共享模式    static final Node SHARED = new Node();    // 标识一个节点是互斥模式    static final Node EXCLUSIVE = null;
    // 标识线程已取消    static final int CANCELLED =  1;    // 标识后继节点需要唤醒    static final int SIGNAL    = -1;    // 标识线程等待在一个条件上    static final int CONDITION = -2;    // 标识后面的共享锁需要无条件的传播（共享锁需要连续唤醒读的线程）    static final int PROPAGATE = -3;
    // 当前节点保存的线程对应的等待状态    volatile int waitStatus;
    // 前一个节点    volatile Node prev;
    // 后一个节点    volatile Node next;
    // 当前节点保存的线程    volatile Thread thread;
    // 下一个等待在条件上的节点（Condition锁时使用）    Node nextWaiter;
    // 是否是共享模式    final boolean isShared() {        return nextWaiter == SHARED;    }
    // 获取前一个节点    final Node predecessor() throws NullPointerException {        Node p = prev;        if (p == null)            throw new NullPointerException();        else            return p;    }
    // 节点的构造方法    Node() {    // Used to establish initial head or SHARED marker    }
    // 节点的构造方法    Node(Thread thread, Node mode) {     // Used by addWaiter        // 把共享模式还是互斥模式存储到nextWaiter这个字段里面了        this.nextWaiter = mode;        this.thread = thread;    }
    // 节点的构造方法    Node(Thread thread, int waitStatus) { // Used by Condition        // 等待的状态，在Condition中使用        this.waitStatus = waitStatus;        this.thread = thread;    }}

// 队列的头节点private transient volatile Node head;// 队列的尾节点private transient volatile Node tail;// 控制加锁解锁的状态变量private volatile int state;

// 获取Unsafe类的实例，注意这种方式仅限于jdk自己使用，普通用户是无法这样调用的private static final Unsafe unsafe = Unsafe.getUnsafe();// 状态变量state的偏移量private static final long stateOffset;// 头节点的偏移量private static final long headOffset;// 尾节点的偏移量private static final long tailOffset;// 等待状态的偏移量（Node的属性）private static final long waitStatusOffset;// 下一个节点的偏移量（Node的属性）private static final long nextOffset;
static {    try {        // 获取state的偏移量        stateOffset = unsafe.objectFieldOffset            (AbstractQueuedSynchronizer.class.getDeclaredField("state"));        // 获取head的偏移量        headOffset = unsafe.objectFieldOffset            (AbstractQueuedSynchronizer.class.getDeclaredField("head"));        // 获取tail的偏移量        tailOffset = unsafe.objectFieldOffset            (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));        // 获取waitStatus的偏移量        waitStatusOffset = unsafe.objectFieldOffset            (Node.class.getDeclaredField("waitStatus"));        // 获取next的偏移量        nextOffset = unsafe.objectFieldOffset            (Node.class.getDeclaredField("next"));
    } catch (Exception ex) { throw new Error(ex); }}
// 调用Unsafe的方法原子更新stateprotected final boolean compareAndSetState(int expect, int update) {    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);}

// 互斥模式下使用：尝试获取锁protected boolean tryAcquire(int arg) {    throw new UnsupportedOperationException();}// 互斥模式下使用：尝试释放锁protected boolean tryRelease(int arg) {    throw new UnsupportedOperationException();}// 共享模式下使用：尝试获取锁protected int tryAcquireShared(int arg) {    throw new UnsupportedOperationException();}// 共享模式下使用：尝试释放锁protected boolean tryReleaseShared(int arg) {    throw new UnsupportedOperationException();}// 如果当前线程独占着锁，返回trueprotected boolean isHeldExclusively() {    throw new UnsupportedOperationException();}

public class MyLockBaseOnAqs {
    // 定义一个同步器，实现AQS类    private static class Sync extends AbstractQueuedSynchronizer {        // 实现tryAcquire(acquires)方法        @Override        public boolean tryAcquire(int acquires) {            if (compareAndSetState(0, 1)) {                setExclusiveOwnerThread(Thread.currentThread());                return true;            }            return false;        }        // 实现tryRelease(releases)方法        @Override        protected boolean tryRelease(int releases) {            setExclusiveOwnerThread(null);            setState(0);            return true;        }    }
    // 声明同步器    private final Sync sync = new Sync();
    // 加锁    public void lock() {        sync.acquire(1);    }
    // 解锁    public void unlock() {        sync.release(1);    }

    private static int count = 0;
    public static void main(String[] args) throws InterruptedException {        MyLockBaseOnAqs lock = new MyLockBaseOnAqs();
        CountDownLatch countDownLatch = new CountDownLatch(1000);
        IntStream.range(0, 1000).forEach(i -> new Thread(() -> {            lock.lock();
            try {                IntStream.range(0, 10000).forEach(j -> {                    count++;                });            } finally {                lock.unlock();            }//            System.out.println(Thread.currentThread().getName());            countDownLatch.countDown();        }, "tt-" + i).start());
        countDownLatch.await();
        System.out.println(count);    }}