# AQS 工具类分析

# 1. CountDownLatch

# API

state: 需要等待的线程数

method:

// 多种实现
public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
public void countDown() {
    sync.releaseShared(1);
}

await：阻塞线程直到latch的数量为0
countDown：释放一个latch

使用的AQS的共享式队列。API支持中断，await的重载方法支持超时。

# 实现

一个静态final内部类实现Sync，维护AQS的state。

state表示还没有完成的任务数。

初始化为固定count，release时每次减1, 实现方法:

protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    for (;;) {
        int c = getState();
        if (c == 0)
            return false;
        int nextc = c-1;
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

API只要使用await进行阻塞，countDown进行释放，state=0会唤醒全部阻塞的线程

# 2. Semaphore

Semaphore持有一定数量的permits，每次调用acquire()都将消耗一个许可，如果许不可用，则阻塞等待直到有许可被释放，每次调用release()都将归还一个许可。通常用来限制资源的访问次数。

Semaphore使用一个抽象的静态内部类Sync实现AQS，实现公共处理逻辑，然后使用NonfairSync和FairSync分别实现公平和非公平的同步器。

state表示许可的数量，每次acquire会减少，每次release会增加

通过实现AQS的tryAcquireShared和tryReleaseShared进行操作

NonfairSync的实现：

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}

// 父类Sync的实现：
final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

可见获取锁就是从state减去个数

释放在Sync进行实现，公平和非公平模式都是同样的。


protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        int current = getState();
        int next = current + releases;
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        if (compareAndSetState(current, next))
            return true;
    }
}

公平模式

protected int tryAcquireShared(int acquires) {
    for (;;) {
        if (hasQueuedPredecessors())
            return -1;
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

多一个判断是否有排队。

Semaphore有个特点可以修改permit，即增加，减少，清空三种操作，其中增加利用Semaphore没有owner的特性，任何线程都可以调用release方法，减少和清空在Sync提供了方法

final void reducePermits(int reductions) {
    for (;;) {
        int current = getState();
        int next = current - reductions;
        if (next > current) // underflow
            throw new Error("Permit count underflow");
        if (compareAndSetState(current, next))
            return;
    }
}

final int drainPermits() {
    for (;;) {
        int current = getState();
        if (current == 0 || compareAndSetState(current, 0))
            return current;
    }
}

增加需要注意溢出的异常，减少注意为负值（溢出）的异常

# 应用示例

ConcurrentHashMap的操作原子性实现并发修改

private static Map<String, Integer> counts = new ConcurrentHashMap<String, Integer>();
/**
 * 将指定的key值加1,借助数据结构的原子操作实现并发修改
 * @param specifiedKey
 */
public void increase(String specifiedKey) {
	while (true) {
		Integer count = counts.get(specifiedKey);
		if (null == count) {
			if (counts.putIfAbsent(specifiedKey, 1) == null) {
				break;
			}
		} else {
			if (counts.replace(specifiedKey, count, count + 1)) {
				break;
			}
		}
	}
}

# CyclicBarrier

用于等待多个线程都准备好，每一个线程准备好后计数器加1 加到指定值后全部开始