【多线程】Semaphore实现原理

「爱情、让人受尽委屈。」 2023-02-10 05:04 25阅读 0赞

## 前言 ##

Semaphore，信号量，一般用于控制同时访问资源的线程数量。可以认为Synchronized代表的是一把锁，那么Semaphore就是多把锁。

--------------------

## 常用方法 ##

public class Semaphore implements java.io.Serializable {
        //构造方法，传入令牌数，默认实例化一个非公平锁
        public Semaphore(int permits);
        //获取一个令牌，在获取成功之前，以及被其他线程中断之前，当前线程会被阻塞
        public void acquire() throws InterruptedException;
        //获取一个令牌，在获取成功之前，当前线程会被阻塞（中断被忽略）
        public void acquireUninterruptibly() ;
        //尝试获取令牌，立即返回获取成功与否，不阻塞当前线程
        public boolean tryAcquire();
        //释放一个令牌
        public void release();
        //返回当前可用的令牌数
        public int availablePermits();
    }

现在有这样的一个例子：

某卫生间只有3个坑位，把坑前面的挡门理解为令牌，因此这里有3个令牌，现在模拟5个人抢坑位的场景。

package com.xue.testSemaphore;
    
    import java.util.concurrent.Semaphore;
    
    public class Main {
        public static void main(String[] args) {
            //最多支持3个人同时蹲坑
            Semaphore semaphore = new Semaphore(3);
            //5个人来抢坑位
            for (int i = 0; i < 5; i++) {
                new Thread(() -> {
                    try {
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName() + "已经在蹲坑");
                        //模拟蹲坑时长
                        Thread.sleep((long) (Math.random() * 10 * 1000));
                        //离开坑位
                        System.out.println(Thread.currentThread().getName() + "即将离开坑位");
                        semaphore.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }, i + "号").start();
            }
        }
    }

输出如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70][]

首先0、1、2号已经抢完了所有的坑位，3与4号只能在外面等候，对的他们没排队（默认实例化了一个非公平锁）。2号出来后，3号才能进去。接着0号出来，4号才能进去。

这个例子虽然有点俗，这确实能让人印象深刻呀。

--------------------

## 原理解析 ##

### 类图 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70 1][]

Semaphore有2个内部类，FairSync与NonfairSync，它们都继承自Sync，Sync又继承自AQS。可以看的出来，Semaphore与CountDownLatch的结构类似，都需要借助于AQS。

对CountDownLatch不熟悉的同学，可以先参考我的另外一篇文章[CountDownLatch实现原理][CountDownLatch]

### 构造方法 ###

public Semaphore(int permits) {
            sync = new NonfairSync(permits);
        }
    
        public Semaphore(int permits, boolean fair) {
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }

默认实例化了一个非公平锁，当然也可以进行指定。这里的permits最终会传到AQS的state变量中，代表当前可用的令牌数。

### acquire() ###

获取一个令牌，获取到线程可以继续执行，否则将会被阻塞。

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

调用了AQS中的模版方法

public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            //尝试获取arg个令牌，该方法返回可用令牌数-需求数，如果小于0，则进行阻塞
            if (tryAcquireShared(arg) < 0)
                doAcquireSharedInterruptibly(arg);
        }

其中tryAcquireShared()由具体的子类(AQS的子类Sync的子类NonfailSync)进行实现

protected int tryAcquireShared(int acquires) {
                return nonfairTryAcquireShared(acquires);
            }

这里又调用了父类Sync的方法

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

remaining =可用令牌数-需求数<0时，直接返回remaining 。否则利用CAS进行更新，同样返回remaining 。对CAS机制不熟悉的同学，可以先参考我的另外一篇文章[浅探CAS实现原理][CAS]

该方法返回一个小于0的值时，将会调用以下方法，这段代码的作用主要就是将获取不到令牌的线程封装为节点，加入到阻塞队列中。

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

### release() ###

释放一个令牌，接着唤醒所有同步队列中的阻塞的共享模式的节点线程。被唤醒的线程重新尝试去获取令牌，获取成功则继续执行，否则重新加入到阻塞队列中。

public void release() {
            sync.releaseShared(1);
        }

releaseShared是AQS中的模版方法

public final boolean releaseShared(int arg) {
            if (tryReleaseShared(arg)) {
                doReleaseShared();
                return true;
            }
            return false;
        }

调用了Sync中的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+循环的方式，将可用令牌数+1。更新成功后，返回ture，最后调用AQS中的doReleaseShared方法

private void doReleaseShared() {
            for (;;) {
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            
                        unparkSuccessor(h);
                    }
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                
                }
                if (h == head)                   
                    break;
            }
        }

该方法会唤醒同步队列中所有被阻塞的共享模式的节点。

关于AQS中方法的详细解释，可能会开另外的篇幅。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200520102544873.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200520104725436.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz,size_16,color_FFFFFF,t_70
[CountDownLatch]: https://blog.csdn.net/qq_33591903/article/details/106135344
[CAS]: https://blog.csdn.net/qq_33591903/article/details/106114086