Java并发之semaphore(信号量)

1:官方解读

semaphore信号量就是并发工具类,Semaphore管理着一组许可permit，许可的初始数量通过构造函数设定。
当线程要访问共享资源时，需要先通过acquire()方法获取许可。获取到之后许可就被当前线程占用了，在归还许可之前其他线程不能获取这个许可。调用acquire()方法时，如果没有许可可用了，就将线程阻塞，等待有许可被归还了再执行。
当执行完业务功能后，需要通过release()方法将许可证归还，以便其他线程能够获得许可证继续执行。

2:通俗易懂的例子解析

我们假设停车场仅有3个停车位，停车位就是有限的共享资源，许可数为3。一开始停车场没有车辆所有车位全部空着，然后先后到来三辆车，停车场车位够，安排进去停车。之后来的车必须在外面候着，直到停车场有空车位。当停车场有车开出去，里面有空位了，则安排一辆车进去（至于是哪辆要看选择的机制是公平还是非公平）。

从程序角度看，停车场就相当于有限的公共资源，许可数为3，车辆就相当于线程。当来一辆车时，许可数就会减1，当停车场没有车位了（许可数为0），其他来的车辆需要在外面等候着。如果有一辆车开出停车场，许可数+1，然后放进来一辆车。

3:代码解析

 public static void main(String[] args) {

        Semaphore semaphore = new Semaphore(3);

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

            new Thread(()->{
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            try {
                System.out.println(Thread.currentThread().getName()+"开始执行");
                Thread.sleep(1000);
                System.out.println(Thread.currentThread().getName()+"执行完毕");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
            }).start();
        }

    }

4:Semaphore的应用

Semaphore可以用于做流量控制，特别是公共资源有限的应用场景，比如数据库连接。假如有多个线程读取数据后，需要将数据保存在数据库中，而可用的最大数据库连接只有10个，这时候就需要使用Semaphore来控制能够并发访问到数据库连接资源的线程个数最多只有10个。在限制资源使用的应用场景下，Semaphore是特别合适的。

5:类结构和相关方法

(1):类结构

Semaphore同样是由AQS实现的，用内部类Sync来管理锁，Sync有两个实现，分别为NonfairSync（非公平锁）和FairSync（公平锁）。
这个类结构有没有似曾相识的感觉，重入锁ReentrantLock也是同样的类结构，Semaphore的源码跟ReentrantLock有很多相似但又比ReentrantLock简单。

(2):acquire()方法

acquire()方法就是获取许可，获取到许可就可以继续执行访问共享资源，获取不到就阻塞等待其他线程归还许可。

入队操作

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

(3):release()方法

• release()方法归还许可，其实就是将AQS.state加1。归还成功，唤醒AQS队列中等锁的线程，从被阻塞的位置开始执行。

 public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {//释放放完毕后  
            doReleaseShared();// 要去唤醒新的结点线程
            return true;
        }
        return false;
    }

• 去释放结点

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;
        }
    }

• 去唤醒结点

private void doReleaseShared() {
   
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;//获取结点的状态  为-1表示的是等待状态
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

6:总结

信号量Semaphore用于控制资源能够被并发访问的线程数量，以保证多个线程能够合理的使用特定资源，比如数据库连接等。

Semaphore在构造时设置一个许可数量，这个许可数量用AQS.state来记录。

acquire()方法就是获取许可，只有获取到许可才可以继续执行访问共享资源，获取到许可之后AQS.state减1，以记录当前可用的许可数量；如果获取不到许可，线程就阻塞等待其他线程归还许可。

release()方法将许可归还，AQS.state加1；归还之后，唤醒AQS队列中阻塞的线程获取许可。