多线程(十三) 同步工具类-Semaphore

public class Test {

    private static Semaphore semaphore = new Semaphore(2);

    public static void main(String[] args) {

        for(int i =1; i <= 10; i++){

            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    select();
                }
            });

            thread.setName("线程序号" + i);
            thread.start();
        }
    }

    private static Object select(){

        try {

            // 注册
            semaphore.acquire();

            // 模拟查询耗时
            Thread.currentThread().sleep(500L);

            // 打印信息
            StringBuilder info = new StringBuilder();
            info.append(Thread.currentThread().getName());
            info.append("进入查询方法,");
            info.append("空闲通道:").append(semaphore.drainPermits());
            info.append(",");
            info.append("等待线程数:").append(semaphore.getQueueLength());
            System.out.println(info.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            // 释放
            semaphore.release();
        }

        return new Object();
    }
}

// 默认使用非公平锁
public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

// 可以通过构造器参数指定是否公平竞争
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

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

    // 初始化访问
    Sync(int permits) {
        setState(permits);
    }

    final int getPermits() {
        return getState();
    }

    // 非公平方式获取AQS共享式资源
    final int nonfairTryAcquireShared(int acquires) {
        // 自旋
        for (;;) {
            // 获取state值
            int available = getState();
            // 计算获取资源后值应该是多少
            int remaining = available - acquires;
            // 如果大于等于0说明满足条件，将计算后值通过CAS修改后返回，如果小于0直接返回
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

    // 释放锁，就是把state加回来
    protected final boolean tryReleaseShared(int releases) {
        for (;;) {
            // 获取state值
            int current = getState();
            // 计算加后的值
            int next = current + releases;
            // 如果加后值小于当前state值，说明参数为负数，抛异常
            if (next < current) // overflow
                throw new Error("Maximum permit count exceeded");
            // 使用CAS方式修改值
            if (compareAndSetState(current, next))
                return true;
        }
    }

    // state减操作
    final void reducePermits(int reductions) {
        // 自旋
        for (;;) {
            // 获取state值
            int current = getState();
            // 计算减后的值
            int next = current - reductions;
            // 如果减后值大于当前state值，说明参数为负数，抛异常
            if (next > current) // underflow
                throw new Error("Permit count underflow");
            // 使用CAS方式修改值
            if (compareAndSetState(current, next))
                return;
        }
    }

    // 将state归零
    final int drainPermits() {

        // 自旋
        for (;;) {

            // 获取state
            int current = getState();

            // 如果是0直接返回0，不是0使用CAS设置成0在返回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);
    }

    // 无限自旋直到CAS修改成功
    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;
        }
    }
}

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

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {

    // 先判断是否已经被中断        
    if (Thread.interrupted())
        throw new InterruptedException();

    // 如果访问次数已经耗尽，进入doAcquireSharedInterruptibly()方法阻塞
    if (tryAcquireShared(arg) < 0)

        /*
         * 源码就不贴了，AQS写好的方法:
         * 排在等待队列的第一个，自旋等待直到重写方法tryAcquireShared(arg)返回值大于0跳出自旋
         * 排在等待队列的第二个开始，直接挂起一边呆着去...
         */
        doAcquireSharedInterruptibly(arg);
}