信号量Semaphore实现原理

　　semaphore用于管理信号量，在并发编程中，可以控制返访问同步代码的线程数量。semaphore在实例化时传入一个int值，也就是指明信号数量。主要方法有两个：acquire()和release()。acquire()用于请求信号，每调用一次，信号量便少一个。release()用于释放信号，调用一次信号量加一个。信号量用完以后，后续使用acquire()方法请求信号的线程便会加入阻塞队列挂起。本篇简单分析semaphore的源码，说明其实现原理。

　　semaphore对于信号量的控制是基于aqs（abstractqueuedsynchronizer）来做的。semaphore有一个内部类sync继承了aqs。而且semaphore中还有两个内部类fairsync和nonfairsync继承sync，也就是说semaphore有公平锁和非公平锁之分。以下是semaphore中内部类的结构：

　　　　

　　看一下semaphore的两个构造函数：

public semaphore(int permits) {
        sync = new nonfairsync(permits);
    }
public semaphore(int permits, boolean fair) {
        sync = fair ? new fairsync(permits) : new nonfairsync(permits);
    }

　　默认是非公平锁。两个构造方法都必须传int permits值。

　　

　　这个int值在实例化内部类时，被设置为aqs中的state。

sync(int permits) {
            setstate(permits);
        }

 

一、acquire()获取信号

　　内部类sync调用aqs中的acquiresharedinterruptibly()方法

public final void acquiresharedinterruptibly(int arg)
            throws interruptedexception {
        if (thread.interrupted())
            throw new interruptedexception();
        if (tryacquireshared(arg) < 0)
            doacquiresharedinterruptibly(arg);
    }

• 调用tryacquireshared()方法尝试获取信号。
• 如果没有可用信号，将当前线程加入等待队列并挂起

　　tryacquireshared()方法被semaphore的内部类nonfairsync和fairsync重写，实现有一些区别。

　　nonfairsync.tryacquireshared()

final int nonfairtryacquireshared(int acquires) {
            for (;;) {
                int available = getstate();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareandsetstate(available, remaining))
                    return remaining;
            }
        }

　　可以看到，非公平锁对于信号的获取是直接使用cas进行尝试的。

 

　　fairsync.tryacquireshared()

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

• 先调用hasqueuedpredecessors()方法，判断队列中是否有等待线程。如果有，直接返回-1，表示没有可用信号
• 队列中没有等待线程，再使用cas尝试更新state，获取信号

　　再看看acquiresharedinterruptibly()方法中，如果没有可用信号加入队列的方法doacquiresharedinterruptibly()

private void doacquiresharedinterruptibly(int arg)
        throws interruptedexception {
        final node node = addwaiter(node.shared);   // 1
        boolean failed = true;
        try {
            for (;;) {
                final node p = node.predecessor();   
                if (p == head) {      // 2
                    int r = tryacquireshared(arg);
                    if (r >= 0) {
                        setheadandpropagate(node, r);
                        p.next = null; // help gc
                        failed = false;
                        return;
                    }
                }
                if (shouldparkafterfailedacquire(p, node) &&     // 3
                    parkandcheckinterrupt())
                    throw new interruptedexception();
            }
        } finally {
            if (failed)
                cancelacquire(node);   
        }
    }

1. 封装一个node节点，加入队列尾部
2. 在无限循环中，如果当前节点是头节点，就尝试获取信号
3. 不是头节点，在经过节点状态判断后，挂起当前线程

二、release()释放信号　　

public final boolean releaseshared(int arg) {
        if (tryreleaseshared(arg)) {    // 1
            doreleaseshared();  // 2
            return true;
        }
        return false;
    }

1. 更新state加一
2. 唤醒等待队列头节点线程

　　tryreleaseshared()方法在内部类sync中被重写

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算法，将state也就是可用信号，加1。

　　

看看semaphore具体的使用示例：

public static void main(string[] args) {
        threadpoolexecutor threadpool = new threadpoolexecutor(10, 10,
                0l, timeunit.milliseconds,
                new linkedblockingqueue<runnable>(10));
        //信号总数为5
        semaphore semaphore = new semaphore(5);
        //运行10个线程
        for (int i = 0; i < 10; i++) {
            threadpool.execute(new runnable() {
                
                @override
                public void run() {
                    try {
                        //获取信号
                        semaphore.acquire();   
                        system.out.println(thread.currentthread().getname() + "获得了信号量,时间为" + system.currenttimemillis());
                        //阻塞2秒，测试效果
                        thread.sleep(2000);
                        system.out.println(thread.currentthread().getname() + "释放了信号量,时间为" + system.currenttimemillis());
                    } catch (interruptedexception e) {
                        e.printstacktrace();
                    } finally {
                        //释放信号
                        semaphore.release();
                    }
                
                }
            });
        }
        threadpool.shutdown();
    }

　　代码结果为：

pool-1-thread-2获得了信号量,时间为1550584196125
pool-1-thread-1获得了信号量,时间为1550584196125
pool-1-thread-3获得了信号量,时间为1550584196125
pool-1-thread-4获得了信号量,时间为1550584196126
pool-1-thread-5获得了信号量,时间为1550584196127
pool-1-thread-2释放了信号量,时间为1550584198126
pool-1-thread-3释放了信号量,时间为1550584198126
pool-1-thread-4释放了信号量,时间为1550584198126
pool-1-thread-6获得了信号量,时间为1550584198126
pool-1-thread-9获得了信号量,时间为1550584198126
pool-1-thread-8获得了信号量,时间为1550584198126
pool-1-thread-1释放了信号量,时间为1550584198126
pool-1-thread-10获得了信号量,时间为1550584198126
pool-1-thread-5释放了信号量,时间为1550584198127
pool-1-thread-7获得了信号量,时间为1550584198127
pool-1-thread-6释放了信号量,时间为1550584200126
pool-1-thread-8释放了信号量,时间为1550584200126
pool-1-thread-10释放了信号量,时间为1550584200126
pool-1-thread-9释放了信号量,时间为1550584200126
pool-1-thread-7释放了信号量,时间为1550584200127

　　可以看到，最多5个线程获得信号，其它线程必须等待获得信号的线程释放信号。