AQS

在线文档-jdk-zh

OSCHINA.NET在线API文档，提供Java API、PHP文档、C语言参考、C++参考、JQuery、Scala等十几种语言的一百多种文档。

https://tool.oschina.net/apidocs/apidoc?api=jdk-zh

Java Platform SE 8

JavaScript is disabled on your browser. This document is designed to be viewed using the frames feature. If you see this message, you are using a non-frame-capable web client. Link to Non-frame version.

https://docs.oracle.com/javase/8/docs/api/index.html

概述

• 用 state 属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁
• getState - 获取 state 状态
• setState - 设置 state 状态
• compareAndSetState - cas 机制设置 state 状态
• 独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源
• 提供了基于 FIFO 的等待队列，类似于 Monitor 的 EntryList
• 条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 Monitor 的 WaitSet
 

• tryAcquire

• tryRelease

• tryAcquireShared

• tryReleaseShared

• isHeldExclusively

//如果获取锁失败
if(!tryAcquire(arg)){
    //入队,可以选择阻塞当前线程 park unpark
}

//如果释放锁成功
if(tryRelease(arg)){
    //让阻塞线程恢复运行
}

自定义不可重入锁

// 自定义锁（不可重入锁）
class MyLock implements Lock {

    // 独占锁  同步器类
    class MySync extends AbstractQueuedSynchronizer {
        @Override
        protected boolean tryAcquire(int arg) {
            if(compareAndSetState(0, 1)) {
                // 加上了锁，并设置 owner 为当前线程
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        @Override // 是否持有独占锁
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        public Condition newCondition() {
            return new ConditionObject();
        }
    }

    private MySync sync = new MySync();

    @Override // 加锁（不成功会进入等待队列）
    public void lock() {
        sync.acquire(1);
    }

    @Override // 加锁，可打断
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    @Override // 尝试加锁（一次）
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    @Override // 尝试加锁，带超时
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(time));
    }

    @Override // 解锁
    public void unlock() {
        sync.release(1);
    }

    @Override // 创建条件变量
    public Condition newCondition() {
        return sync.newCondition();
    }
}

public class TestAqs {
    public static void main(String[] args) {
        MyLock lock = new MyLock();
        new Thread(() -> {
            lock.lock();
            try {
                log.debug("locking...");
                sleep(1);
            } finally {
                log.debug("unlocking...");
                lock.unlock();
            }
        },"t1").start();

        new Thread(() -> {
            lock.lock();
            try {
                log.debug("locking...");
            } finally {
                log.debug("unlocking...");
                lock.unlock();
            }
        },"t2").start();
    }
}     
//15:34:42.708 c.TestAqs [t1] - locking...
//15:34:43.723 c.TestAqs [t1] - unlocking...
//15:34:43.723 c.TestAqs [t2] - locking...
//15:34:43.723 c.TestAqs [t2] - unlocking...

设计思想

• 获取锁超时机制

• 阻塞版本获取锁 acquire 和非阻塞的版本尝试获取锁 tryAcquire

• 通过打断取消机制

• 独占机制及共享机制

• 条件不满足时的等待机制

while(state 状态不允许获取) {
	if(队列中还没有此线程) {
	入队并阻塞
	}
}
当前线程出队

if(state 状态允许了) {
	恢复阻塞的线程(s)
}

• 原子维护 state 状态

• 阻塞及恢复线程

• 维护队列

• state 使用 volatile 配合 cas 保证其修改时的原子性

• state 使用了 32bit int 来维护同步状态，因为当时使用 long 在很多平台下测试的结果并不理想

• 早期的控制线程暂停和恢复的 api 有 suspend 和 resume，但它们是不可用的，因为如果先调用的 resume那么 suspend 将感知不到

• 解决方法是使用 park & unpark 来实现线程的暂停和恢复，先 unpark 再 park 也没问题

• park & unpark 是针对线程的，而不是针对同步器的，因此控制粒度更为精细

• park 线程还可以通过 interrupt 打断

• 使用了 FIFO 先入先出队列，并不支持优先级队列

• 设计时借鉴了 CLH 队列，它是一种单向无锁队列
• 队列中有 head 和 tail 两个指针节点，都用 volatile 修饰配合 cas 使用，每个节点有 state 维护节点状态

do {
	// 原来的 tail
	Node prev = tail;
	// 用 cas 在原来 tail 的基础上改为 node
} while(tail.compareAndSet(prev, node))

// prev 是上一个节点
while((Node prev=node.prev).state != 唤醒状态) {
}
// 设置头节点
head = node;

• 无锁，使用自旋

• 快速，无阻塞

private Node enq(final Node node) {
	for (;;) {
		Node t = tail;
		// 队列中还没有元素 tail 为 null
		if (t == null) {
		// 将 head 从 null -> dummy
		if (compareAndSetHead(new Node()))
			tail = head;
		} else {
		// 将 node 的 prev 设置为原来的 tail
			node.prev = t;
		// 将 tail 从原来的 tail 设置为 node
			if (compareAndSetTail(t, node)) {
			// 原来 tail 的 next 设置为 node
			t.next = node;
			return t;
			}
		}	
	}
}