Semaphore

• Semaphore
  • 1、Semaphore概念
    • 1.1、PV操作
  • 2、Semaphore 常用方法
    • 2.1、构造器确定资源数目
  • 3、常用方法
  • 4、应用场景
  • 5、源码分析
    • 5.1、acquire
    • 5.2、release
  • 6、总结
    • 6.1、DEMO1
    • 6.2、DEMO2

1、Semaphore概念

Semaphore，俗称信号量，它是操作系统中PV操作的原语在java的实现，它也是基于AbstractQueuedSynchronizer实现的。

Semaphore的功能非常强大，大小为1的信号量就类似于互斥锁，通过同时只能有一个线程获取信号量实现。大小为n（n>0）的信号量可以实现限流的功能，它可以实现只能有n个线程同时获取信号量。

1.1、PV操作

PV操作是操作系统一种实现进程互斥与同步的有效方法。PV操作与信号量（S）的处理相关，P表示通过的意思，V表示释放的意思。用PV操作来管理共享资源时，首先要确保PV操作自身执行的正确性。

P操作的主要动作是：

①S减1；

②若S减1后仍大于或等于0，则进程继续执行；

③若S减1后小于0，则该进程被阻塞后放入等待该信号量的等待队列中，然后转进程调度。

V操作的主要动作是：

①S加1；

②若相加后结果大于0，则进程继续执行；

③若相加后结果小于或等于0，则从该信号的等待队列中释放一个等待进程，然后再返回原进程继续执行或转进程调度。

2、Semaphore 常用方法

2.1、构造器确定资源数目

permits 表示许可证的数量（资源数）

fair 表示公平性，如果这个设为 true 的话，下次执行的线程会是等待最久的线程 。

    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }


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

查看源码进去之后，可以看到最终到了：

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

可以看到传入进去的参数将会作为资源数目的多少来进行定义。

3、常用方法

public void acquire() throws InterruptedException
public boolean tryAcquire()
public void release()
public int availablePermits()
public final int getQueueLength()
public final boolean hasQueuedThreads()
protected void reducePermits(int reduction)
protected Collection getQueuedThreads()

4、应用场景

可以用于做流量控制，特别是公用资源有限的应用场景

示例：

/**
 * @Description 有三个许可，五个线程来抢。效果：三个线程执行成功，另外两个线程进行排队
 * @Author liguang
 * @Date 2022/03/18/11:13
 */
public class SemphoreTestOne {
    public static void main(String[] args) {
        Semaphore windows = new Semaphore(3);
        for (int i = 0; i < 5; i++) {
            new Thread(()->{
                try {
                    windows.acquire();
                    System.out.println("开始卖票");
                    Thread.sleep(5000);
                    System.out.println("购票成功");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                  windows.release();
                }
            }).start();
        }
    }
}

控制台展示：

开始卖票
开始卖票
开始卖票
-----------------停顿5s钟----------------    
购票成功
购票成功
开始卖票
开始卖票
购票成功

效果上：许可只有三个，但是有多个线程，那么多个线程来获取得到三个有限的资源的时候，对于已经获取得到了的线程来说，是没有任何影响，而没有获取得到的线程只能够排队等待资源。

5、源码分析

这里将会涉及到共享锁的一些概念。

从两行代码中来进行分析：

 Semaphore windows = new Semaphore(3);
try {
    windows.acquire();
    xxxxx(业务方法);
} catch (InterruptedException e) {
    e.printStackTrace();
} finally {
    windows.release();
}

那么在acquire和release就可以实现这样的逻辑，那么看下具体的实现逻辑。

5.1、acquire

只需要来分析这段代码即可：

        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);

那么看下尝试获取得到共享锁，当对应的值小于0的时候，才会去将线程进行阻塞。

首先看下尝试获取得到共享锁的方法（非公平场景下来进行分析）：

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        // 1、当许可的数量小于0（表示的当前资源不足）；2、大于等于0的时候，表示资源还是足够的，可以分配给当前线程
        if (remaining < 0 || compareAndSetState(available, remaining))
            // 许可剩余数量
            return remaining;
    }
}

对于没有获取得到成功的，那么将继续循环操作。最终将返回许可的剩余数量。

那么在回到上面的方法中来，只有当许可小于0的时候，才会将对应的线程进行阻塞。而大于等于0的，则是直接获取得到资源进行操作了

    private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        // 这里和之前的还是有区别的，这里使用到节点的另外一种状态。SHARED，共享状态。构建一个共享队列来进行创建
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    // 如果是头节点！那么这里再来尝试获取得到一次锁，如果得到的
                    int r = tryAcquireShared(arg);
                    // 如果获取得到的资源大于等于0，那么将会进行下面的操作
                    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);
        }
    }

那么下面只需要关注一下setHeadAndPropagate方法即可

    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head;
        setHead(node);
        if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            // 在这里要来唤醒下一个节点。这里直接在这里来进行唤醒了！
            if (s == null || s.isShared())
                // 可以看到释放锁也是在这里来进行操作的。在这里直接来通知后面的排队的线程来进行操作。
                // 唤醒后面的线程来抢锁
                doReleaseShared();
        }
    }

5.2、release

释放也是非常简单的，直接上源码：

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

直接获取得到原来的值，然后进行相加重新设置+1即可。

看看释放锁的过程：

    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;            // 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、总结

这个源码比较简单，所以不过来过多解释。无非是检测到了状态是SHARED并且不为null的，直接进行唤醒来尝试通知下一个线程即可。

但是使用的时候需要注意的一点是：要有效率的来获取得到许可数量。

6.1、DEMO1

线程获取得到许可证的数量不能够大于许可证原始数量

如下所示：

/**
 * @Description 获取得到许可过大。注意这里的许可数量设置
 * @Author liguang
 * @Date 2022/03/18/11:13
 */
public class SemphoreTestTwo {
    public static void main(String[] args) {
        
        Semaphore windows = new Semaphore(5);
        for (int i = 0; i < 5; i++) {
            new Thread(()->{
                try {
                    windows.acquire(6);
                    System.out.println("开始卖票");
                    Thread.sleep(5000);
                    System.out.println("购票成功");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                  windows.release();
                }
            }).start();
        }
    }
}

这种操作最终将会导致所有的线程都将会在阻塞队列中进行排队，等待被激活。无法被唤醒，除非是因为操作不当，因为中断引起。

6.2、DEMO2

线程获取得到多少许可证，在释放的时候就要释放多少许可证

再来一个例子：

/**
 * @Description 获取得到许可证的的数量要和释放许可证的数量保持一致
 * @Author liguang
 * @Date 2022/03/18/11:13
 */
public class SemphoreTestTwo {
    public static void main(String[] args) {
        Semaphore windows = new Semaphore(5);
        for (int i = 0; i < 5; i++) {
            new Thread(()->{
                try {
                    windows.acquire(3);
                    System.out.println("开始卖票");
                    Thread.sleep(5000);
                    System.out.println("购票成功");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                  windows.release(3);
                }
            }).start();
        }
    }
}