Semaphore实现原理分析

Posted 2020-10-06 大诚挚

synchronized的语义是互斥锁，就是在同一时刻，只有一个线程能获得执行代码的锁。但是现实生活中，有好多的场景，锁不止一把。

比如说，又到了十一假期，买票是重点，必须圈起来。在购票大厅里，有5个售票窗口，也就是说同一时刻可以服务5个人。要实现这种业务需求，用synchronized显然不合适。

查看Java并发工具，发现有一个Semaphore类，天生就是处理这种情况的。

先用Semaphore实现一个购票的小例子，来看看如何使用

 

package semaphore;

import java.util.concurrent.Semaphore;

public class Ticket {

    public static void main(String[] args) {
        Semaphore windows = new Semaphore(5);  // 声明5个窗口
        
        for (int i = 0; i < 8; i++) {
            new Thread() {
                @Override
                public void run() {
                    try {
                        windows.acquire();  // 占用窗口
                        System.out.println(Thread.currentThread().getName() + ": 开始买票");
                        sleep(2000);  // 睡2秒，模拟买票流程
                        System.out.println(Thread.currentThread().getName() + ": 购票成功");
                        windows.release();  // 释放窗口
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }.start();
        }
    }
}

 

运行结果

Thread-1: 开始买票
Thread-3: 开始买票
Thread-4: 开始买票
Thread-2: 开始买票
Thread-0: 开始买票
Thread-1: 购票成功
Thread-5: 开始买票
Thread-3: 购票成功
Thread-2: 购票成功
Thread-4: 购票成功
Thread-7: 开始买票
Thread-6: 开始买票
Thread-0: 购票成功
Thread-7: 购票成功
Thread-5: 购票成功
Thread-6: 购票成功

View Code

从结果来看，最多只有5个线程在购票。而这么精确的控制，我们也只是调用了acquire和release方法。下面看看是如何实现的。

从acquire方法进去，又可以看到老套路：具体调用的还是AbstractQueuedSynchronizer这个类的逻辑

    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

而tryAcquireShared方法留给了子类去实现，Semaphore类里面的两个内部类FairSync和NonfairSync都继承自AbstractQueuedSynchronizer。这两个内部类，从名字来看，一个实现了公平锁，另一个是非公平锁。这里多说一句，所谓公平和非公平是这个意思：假设现在有一个线程A在等待获取锁，这时候又来了线程B，如果这个时候B不考虑A的感受，也去申请锁，显然不公平；反之，只要A是先来的，B一定要排在A的后面，不能马上去申请锁，就是公平的。

Semaphore默认是调用了NonfairSync的tryAcquireShared方法，主要逻辑：

        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

这又是一个经典的CAS操作加无限循环的算法，用来保证共享变量的正确性。另外，此处的getState()方法很是迷惑人，你以为是获取状态，实则不然。我们先看看Semaphore的构造方法：

    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
        // 内部类
        NonfairSync(int permits) {
            super(permits);
        }
        // 内部类，NonfairSync的父类
        Sync(int permits) {
            setState(permits);
        }

我们传进去的参数5，最终传给了setState方法，而getState和setState方法都在AbstractQueuedSynchronizer类里面

    /**
     * The synchronization state.
     */
    private volatile int state;

    protected final int getState() {
        return state;
    }

    protected final void setState(int newState) {
        state = newState;
    }

也就是说父类定义了一个属性state，并配有final的get和set方法，子类只需要继承该属性，想代表什么含义都可以，比如Semaphore里面的内部类Sync就把这个属性当作最大允许访问的permits，像CountDownLatch和CyclicBarrier都是这么干的。这种方式似乎不太好理解，为什么不是每个子类都定义自己的具有明确语义的属性，而是把控制权放在父类？？？我猜是出于安全的考虑。反正，大师的思考深度，我们揣摩不了。

再回到tryAcquireShared方法，这个方法是有参数的---int型的acquires，代表你要一次占几个坑。我们调用的无参的acquire方法，默认是传入1作为参数调用的这个方法，一次只申请一个坑。但是有的情况下，你可能一次需要多个，比如高富帅需要同时交多个女朋友。方法的返回值是剩余的坑的数量，如果数量小于0，执行AbstractQueuedSynchronizer这个类的doAcquireSharedInterruptibly方法。

    /**
     * Acquires in shared interruptible mode.
     * @param arg the acquire argument
     */
    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);
        }
    }

这个方法的逻辑与独占模式下的逻辑差不多，可以看看之前讲Condition的那篇，出门一路左拐。当所有的坑都被占着的时候，再来的线程都会被封装成节点，添加到等待的队列里面去。不同的是，这里的节点都是共享模式，而共享模式是实现多个坑同时提供服务的核心。

再来看看坑的释放，从release方法进去，核心逻辑在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");
                if (compareAndSetState(current, next))
                    return true;
            }
        }

CAS、无限循环，熟悉的配方，熟悉的味道。同获取一样，这里也可以一次释放多个坑。然而，这里考虑到了next小于current的情况，我是绞尽脑汁也没想出来。传进来的releases一般都是大于0的整数（大部分情况下就是1），最终还是会造成next小于current，实在是想不出来，而且还是抛出Error。但是这种情况，通过代码可以精确的再现。好吧，是在下输了。如果读者中有高人，请指点一二，不胜感激！！！

前面这么多都只是分析了非公平模式下的处理逻辑，而公平模式下的逻辑多了一个判断，就是看看前面还有没有线程在等待（节点有没有前驱）。具体的细节，希望读者自己玩味。

最后总结一下：所有的并发核心控制逻辑都在AbstractQueuedSynchronizer这个类中，只有理解了这个类的设计思路，才能真正理解衍生出来的工具类的实现原理。