6.32 非公平信号量

1. 是什么

限流，使用的非公平策略

2. 使用

public class SemaphoreTest
{
    private final static int THREAD_COUNT = 100;
    private final static CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT);

    public static void main(String[] args) throws InterruptedException
    {
        Semaphore semaphore = new Semaphore(10);

        for (int i = 0; i < THREAD_COUNT; i++)
        {
            new Thread(()->{
                try
                {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + "正在访问资源。。。");
                    TimeUnit.SECONDS.sleep(3);

                }
                catch (Exception e)
                {
                    e.printStackTrace();
                }
                finally
                {
                    semaphore.release();
                    countDownLatch.countDown();
                }
            }).start();
        }

        countDownLatch.await();
    }
}

3. 原理分析

3.1. 构造方法

public Semaphore(int permits) {
    //创建的是非公平的Sync
    sync = new NonfairSync(permits);
}

3.1.1. 非公平Sync

• NonfairSync

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

    NonfairSync(int permits) {
        //调用Sync
        super(permits);
    }

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

• Sync

abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 1192457210091910933L;

    Sync(int permits) {
        //最终就是设置了permits个信号量
        setState(permits);
    }
}

3.2. acquire

public void acquire() throws InterruptedException {
    //AQS的acquireSharedInterruptibly
    sync.acquireSharedInterruptibly(1);
}

3.2.1. 调用AQS加共享锁

• AQS acquireSharedInterruptibly

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    //Semaphore.NonfairSync重写了tryAcquireShared
    //如果信号量不够那么返回负数，这里执行doAcquireSharedInterruptibly入AQS队,并且阻塞等待唤醒
    //如果信号量够的话返回>=0的数，那么调用此acquire方法的代码就可以继续执行业务代码
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

3.2.1.1. 尝试加共享锁【非公平：不管队列前面是否有人我直接尝试加锁】

• Semaphore.NonfairSync#tryAcquireShared

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

• Semaphore.Sync#nonfairTryAcquireShared

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        //当前的信号量
        int available = getState();
        //当前的信号量够不够我获取？
        int remaining = available - acquires;
        //<0表示不够的话返回这个数
        if (remaining < 0 ||
        //>=0说明够了，那么CAS修改剩余信号量
            compareAndSetState(available, remaining))
            return remaining;
    }
}

3.3. release

• Semaphore#release()

public void release() {
    //AQS的releaseShared
    sync.releaseShared(1);
}

3.3.1. 调用AQS释放共享锁

• AQS#releaseShared

public final boolean releaseShared(int arg) {
    //Semaphore.Sync重写了tryReleaseShared
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

3.3.1.1. 尝试释放锁

• Semaphore.Sync#tryReleaseShared

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");
        //CAS修改信号量
        if (compareAndSetState(current, next))
            return true;
    }
}