转载

J.U.C 之Semaphore

信号量 Semaphore 是一个控制访问多个共享资源的计数器，和 CountDownLatch 一样，其本质上是一个“共享锁”。

一个计数信号量。从概念上讲，信号量维护了一个许可集。

如有必要，在许可可用前会阻塞每一个 acquire，然后再获取该许可。
每个 release 添加一个许可，从而可能释放一个正在阻塞的获取者。
复制代码

但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。

Semaphore 通常用于限制可以访问某些资源（物理或逻辑的）的线程数目。

下面我们就一个停车场的简单例子来阐述 Semaphore ：

  1. 为了简单起见我们假设停车场仅有 5 个停车位。一开始停车场没有车辆所有车位全部空着，然后先后到来三辆车，停车场车位够，安排进去停车，然后又来三辆，这个时候由于只有两个停车位，所有只能停两辆，其余一辆必须在外面候着，直到停车场有空车位。当然，以后每来一辆都需要在外面候着。当停车场有车开出去，里面有空位了，则安排一辆车进去（至于是哪辆，要看选择的机制是公平还是非公平）。

  2. 从程序角度看，停车场就相当于信号量 Semaphore ，其中许可数为 5 ，车辆就相对线程。当来一辆车时，许可数就会减 1 。当停车场没有车位了（许可数 == 0 ），其他来的车辆需要在外面等候着。如果有一辆车开出停车场，许可数 + 1，然后放进来一辆车。

  3. 信号量 Semaphore 是一个非负整数（ >=1 ）。当一个线程想要访问某个共享资源时，它必须要先获取 Semaphore。当 Semaphore > 0 时，获取该资源并使 Semaphore – 1 。如果S emaphore 值 = 0，则表示全部的共享资源已经被其他线程全部占用，线程必须要等待其他线程释放资源。当线程释放资源时，Semaphore 则 +1 。

实现分析

java.util.concurrent.Semaphore 结构如下图：

J.U.C 之Semaphore

从上图可以看出，Semaphore 内部包含公平锁（FairSync）和非公平锁（NonfairSync），继承内部类 Sync ，其中 Sync 继承 AQS（再一次阐述 AQS 的重要性）。

Semaphore 提供了两个构造函数：

Semaphore(int permits) ：创建具有给定的许可数和非公平的公平设置的 Semaphore 。
Semaphore(int permits, boolean fair) ：创建具有给定的许可数和给定的公平设置的 Semaphore 。
复制代码

实现如下：

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

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

Semaphore 默认选择非公平锁。

当信号量 Semaphore = 1 时，它可以当作互斥锁使用。其中 0、1 就相当于它的状态：

  1. 当 =1 时表示，其他线程可以获取；
  2. 当 =0 时，排他，即其他线程必须要等待。

Semaphore 的代码实现结构，和 ReentrantLock 类似。

信号量获取

Semaphore 提供了 #acquire() 方法，来获取一个许可。

public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
复制代码

内部调用 AQS 的 #acquireSharedInterruptibly(int arg) 方法，该方法以共享模式获取同步状态。代码如下：

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

在 #acquireSharedInterruptibly(int arg) 方法中，会调用 #tryAcquireShared(int arg) 方法。而 #tryAcquireShared(int arg) 方法，由子类来实现。对于 Semaphore 而言，如果我们选择非公平模式，则调用 NonfairSync 的#tryAcquireShared(int arg) 方法，否则调用 FairSync 的 #tryAcquireShared(int arg) 方法。若 #tryAcquireShared(int arg) 方法返回 < 0 时，则会阻塞等待，从而实现 Semaphore 信号量不足时的阻塞，代码如下：

// AQS.java
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;
                }
            }
            /**
             * 对于 Semaphore 而言，如果 tryAcquireShared 返回小于 0 时，则会阻塞等待。
             */
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
复制代码

另外，这也是为什么 Semaphore 在使用 AQS 时，state 代表的是，剩余可获取的许可数，而不是已经使用的许可数。我们假设 state 代表的是已经使用的许可数，那么 #tryAcquireShared(int arg) 返回的结果 = 原始许可数 - state ，这个操作在并发情况下，会存在线程不安全的问题。所以，state 代表的是，剩余可获取的许可数，而不是已经使用的许可数。

公平情况的 FairSync 的方法实现，代码如下：

// FairSync.java
@Override
protected int tryAcquireShared(int acquires) {
    for (;;) {
        //判断该线程是否位于CLH队列的列头，从而实现公平锁
        if (hasQueuedPredecessors())
            return -1;
        //获取当前的信号量许可
        int available = getState();

        //设置“获得acquires个信号量许可之后，剩余的信号量许可数”
        int remaining = available - acquires;

        //CAS设置信号量
        if (remaining < 0 ||
                compareAndSetState(available, remaining))
            return remaining;
    }
}
复制代码

通过 #hasQueuedPredecessors() 方法，判断该线程是否位于 CLH 队列的列头，从而实现公平锁。 非公平情况的 NonfairSync 的方法实现，代码如下：

// NonfairSync.java
protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}

// Sync.java
final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}
复制代码

对于非公平而言，因为它不需要判断当前线程是否位于 CLH 同步队列列头，所以相对而言会简单些。

信号量释放

获取了许可，当用完之后就需要释放，Semaphore 提供 #release() 方法，来释放许可。代码如下：

public void release() {
    sync.releaseShared(1);
}
复制代码

内部调用 AQS 的 #releaseShared(int arg) 方法，释放同步状态。

// AQS.java
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}
复制代码

releaseShared(int arg) 方法，会调用 Semaphore 内部类 Sync 的 #tryReleaseShared(int arg) 方法，释放同步状态。

// Sync.java
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");
        //设置可获取的信号许可数为next
        if (compareAndSetState(current, next))
            return true;
    }
}
复制代码

如该方法返回 true 时，代表释放同步状态成功，从而在 #releaseShared(int args) 方法中，调用 #doReleaseShared() 方法，可唤醒阻塞等待 Semaphore 的许可的线程。

应用示例

我们已停车为示例：

public class SemaphoreTest {

    static class Parking {
    
        //信号量
        private Semaphore semaphore;

        Parking(int count) {
            semaphore = new Semaphore(count);
        }

        public void park() {
            try {
                //获取信号量
                semaphore.acquire();
                long time = (long) (Math.random() * 10);
                System.out.println(Thread.currentThread().getName() + "进入停车场，停车" + time + "秒..." );
                Thread.sleep(time);
                System.out.println(Thread.currentThread().getName() + "开出停车场...");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
        }
    }


    static class Car extends Thread {
        Parking parking ;

        Car(Parking parking){
            this.parking = parking;
        }

        @Override
        public void run() {
            parking.park();     //进入停车场
        }
    }

    public static void main(String[] args){
        Parking parking = new Parking(3);

        for(int i = 0 ; i < 5 ; i++){
            new Car(parking).start();
        }
    }
}
复制代码

运行结果如下：

J.U.C 之Semaphore
原文  https://juejin.im/post/5d8b184151882534b02f17d0
正文到此结束
所属分类： 编程技术 Java

热门推荐

相关文章

阿里云首购8折
说给你听

本文目录
随机标签
书籍教程
springboot-demo Java入门教程 bootstrap3 CSS Apache基础教程 php ionic 教程 Python mysql教程 eclipse Ubuntu VPS系统配置 AngularJS 教程 MongoDB教程 Struts2教程 Redis教程 Spring教程 Git教程 Jenkins进阶系列 openfire参考指南 Java设计模式 HBase教程 Maven教程 hibernate教程 Docker 教程 memcached教程 Quartz指南 springcloud-demo Hive教程 Ant教程 ANTLR教程 SpringCloud java实例教程 XStream教程 Hazelcast教程 Elastic-Job-Lite 深入浅出MyBatis SVN教程 ibaties教程 rabittmq教程 solr教程 Hadoop教程 WebService CXF学习 JPA教程 ActiveMQ中文指南 java-demo Java内存模型 dubbo教程 python3-demo Linux入门视频教程
近期评论
  1. 1 如何将Spring Cloud Task发布Data Flow上执行？
  2. 2 Spring Cloud Data Flow快速入门Demo
  3. 3 Spring Cloud Consul实现选举机制
  4. 4 Spring Boot集成ShedLock实现分布式定时任务
  5. 5 利用oss进行数据库和网站图片备份
  6. 6 Spring Cloud Stream实现数据流处理
  7. 7 Python3访问MySQL数据库快速入门Demo
  8. 8 Github开源项目作者可以免费申请 JetBrains 全家桶
  9. 9 Spring Cloud Vault快速入门Demo
  10. 10 Spring Cloud Gateway快速入门Demo
网站信息
  • 文章总数：82,716 篇
  • 文件总数：284,297 个
  • 标签总数：2,396 个
  • 分类总数：85 个
  • 留言数量：2,560 条
  • 在线人数：699 人
  • 运行天数：4,410天
  • 最后更新：2024年11月23日15点
Loading...

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

[HBLOG]公众号 HBLOG HBLOG