并发工具类Semaphore信号量源码解析

/**
 * 信号量的同步实现  
 * 使用AQS的状态来表示许可
 * 子类分别实现了公平和非公平的两个版本
 */
abstract static class Sync extends AbstractQueuedSynchronizer {
	private static final long serialVersionUID = 1192457210091910933L;
	// 构造函数设置了AQS的状态值
	Sync(int permits) {
		setState(permits);
	}
	// 获取AQS的状态值
	final int getPermits() {
		return getState();
	}
	// 共享模式下以非公平方式获取
	final int nonfairTryAcquireShared(int acquires) {
		for (;;) {
			int available = getState();
			// 设置获取后的剩余通行量
			int remaining = available - acquires;
			// remaining<0时,放到队列中阻塞;remaining>0时,则可以通行
			if (remaining < 0 ||
				compareAndSetState(available, remaining))
				return remaining;
		}
	}
	// 共享模式下进行释放
	protected final boolean tryReleaseShared(int releases) {
		for (;;) {
			int current = getState();
            // 释放通行量,使得可用的通行量增加,别的线程就可以使用
			int next = current + releases;
			if (next < current) // 超出最大允许释放数
				throw new Error("Maximum permit count exceeded");
			// 设置释放后的剩余通行量,返回释放成功,表示需要通知下一个线程通行	
			if (compareAndSetState(current, next)) // 剩下的通行数
				return true;
		}
	}
	// 根据指定的减少量减小可用通行数
	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();
			// 可用通行数设置为0
			if (current == 0 || compareAndSetState(current, 0))
				return current;
		}
	}
}

static final class NonfairSync extends Sync {
	private static final long serialVersionUID = -2694183684443567898L;

	NonfairSync(int permits) {
		super(permits);
	}
	// 共享模式下获取
	protected int tryAcquireShared(int acquires) {
		return nonfairTryAcquireShared(acquires);
	}
}

static final class FairSync extends Sync {
	private static final long serialVersionUID = 2014338818796000944L;

	FairSync(int permits) {
		super(permits);
	}

	protected int tryAcquireShared(int acquires) {
		for (;;) {
			// 如果当前线程不位于队列头部,则阻塞
			if (hasQueuedPredecessors())
				return -1;
			int available = getState();
			int remaining = available - acquires;
			// 设置获取后的剩余通行量,当剩余量小0时,则阻塞,否则通行
			if (remaining < 0 ||
				compareAndSetState(available, remaining))
				return remaining;
		}
	}
}

/** 所有操作都是通过AbstractQueuedSynchronizer子类去实现 */
private final Sync sync;

/**
 * 通过指定的permits参数设置通行量和默认使用非公平方式
 * permits可能为负值,在这种情况下,必须在任何操作前进行释放
 */
public Semaphore(int permits) {
	sync = new NonfairSync(permits);
}

/**
 * 通过指定的permits参数设置通行量和参数来决定公平还是非公平锁
 */
public Semaphore(int permits, boolean fair) {
	sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

/**
 * 可中断的方式从信号量获取一次通行
 * 如果能够获取到通行量,则立即返回,并相应减少一次通行
 * 如果没有可用的通行,则当前线程因线程调度而被禁用,并且在发生以下两种情况之一时处于休眠:
 * 1.其他线程调用信号量的release方法,并且将当前线程指定为下一个被许可的
 * 2.其他线程中断当前线程
 */
public void acquire() throws InterruptedException {
	sync.acquireSharedInterruptibly(1);
}

/**
 * 释放一次通行,将其返回给信号量,会增加通行的数量.
 */
public void release() {
	sync.releaseShared(1);
}

public class SemaphoreDemo {
    public void method(Semaphore semaphore) {
        try {
            System.out.println(Thread.currentThread().getName() + " tryAcquire ..");
            semaphore.acquire(5);
            System.out.println(Thread.currentThread().getName() + " acquire success ..");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release(5);
            System.out.println(Thread.currentThread().getName() + " release..");
        }
    }

    public static void main(String[] args) {
        SemaphoreDemo demo = new SemaphoreDemo();
        Semaphore semaphore = new Semaphore(12);

        for (int i = 0; i < 3; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    demo.method(semaphore);
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}