欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

信号量Semaphore实现原理

程序员文章站 2022-04-23 23:28:11
Semaphore用于管理信号量,在并发编程中,可以控制返访问同步代码的线程数量。Semaphore在实例化时传入一个int值,也就是指明信号数量。主要方法有两个:acquire()和release()。acquire()用于请求信号,每调用一次,信号量便少一个。release()用于释放信号,调用 ......

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

  semaphore对于信号量的控制是基于aqs(abstractqueuedsynchronizer)来做的。semaphore有一个内部类sync继承了aqs。而且semaphore中还有两个内部类fairsync和nonfairsync继承sync,也就是说semaphore有公平锁和非公平锁之分。以下是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个线程获得信号,其它线程必须等待获得信号的线程释放信号。