Semaphore 源码解析

semaphore 实现

使用 Semaphore 限流，在访问高峰期时，让请求线程阻塞，高峰期过去再释放许可，当然它只适合限制单机线程数量，并且仅是限制线程数，而不是限制资源数（例如连接数，请对比 Tomcat LimitLatch 的实现可以限制连接的资源个数） 用 Semaphore 实现简单连接池，对比使用object的wait notify，性能和可读性显然更好，

package com.dongguo.juc;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;


@Slf4j(topic = "d.SemaphoreTest")
public class SemaphoreTest {
    public static void main(String[] args) {
        // 1. 创建 semaphore 对象
        Semaphore semaphore = new Semaphore(3);

        // 2. 10个线程同时运行
        for (int i = 0; i  {
                try {
                    semaphore.acquire();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    log.debug("running...");
                    TimeUnit.SECONDS.sleep(1);
                    log.debug("end...");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release();
                }
            }).start();
        }
    }
}
运行结果
21:10:45 [Thread-0] d.SemaphoreTest - running...
21:10:45 [Thread-1] d.SemaphoreTest - running...
21:10:45 [Thread-2] d.SemaphoreTest - running...
21:10:46 [Thread-1] d.SemaphoreTest - end...
21:10:46 [Thread-3] d.SemaphoreTest - running...
21:10:46 [Thread-0] d.SemaphoreTest - end...
21:10:46 [Thread-8] d.SemaphoreTest - running...
21:10:46 [Thread-2] d.SemaphoreTest - end...
21:10:46 [Thread-9] d.SemaphoreTest - running...
21:10:47 [Thread-3] d.SemaphoreTest - end...
21:10:47 [Thread-8] d.SemaphoreTest - end...
21:10:47 [Thread-6] d.SemaphoreTest - running...
21:10:47 [Thread-7] d.SemaphoreTest - running...
21:10:47 [Thread-9] d.SemaphoreTest - end...
21:10:47 [Thread-5] d.SemaphoreTest - running...
21:10:48 [Thread-7] d.SemaphoreTest - end...
21:10:48 [Thread-6] d.SemaphoreTest - end...
21:10:48 [Thread-4] d.SemaphoreTest - running...
21:10:48 [Thread-5] d.SemaphoreTest - end...
21:10:49 [Thread-4] d.SemaphoreTest - end...

介绍

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

Semaphore通俗理解就是常说的共享锁，是一个控制访问多个共享资源的计数器,它可以定义共享资源的个数，只要共享资源还有，其他线程就可以执行，否则就会被阻塞。 Semaphore，在 API 是这么介绍的： 一个计数信号量。从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire，然后再获取该许可。每个 release 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。

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

public class Semaphore implements java.io.Serializable

常量&变量

    //序列化版本号
    private static final long serialVersionUID = -3222578661600680210L;
    /** All mechanics via AbstractQueuedSynchronizer subclass */
    /** 所有机制通过AbstractQueuedSynchronizer子类实现 */
    private final Sync sync;

构造方法

Semaphore 提供了两个构造函数： 1.Semaphore(int permits) ：创建具有给定的许可数和非公平的公平设置的 Semaphore 。 2.Semaphore(int permits, boolean fair) ：创建具有给定的许可数和给定的公平设置的 Semaphore

    /**
     * Creates a {@code Semaphore} with the given number of
     * permits and nonfair fairness setting.
     *
     * @param permits the initial number of permits available.
     *        This value may be negative, in which case releases
     *        must occur before any acquires will be granted.
     *                指定许可证个数，默认采用非公平版本
     */
    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }

    /**
     * Creates a {@code Semaphore} with the given number of
     * permits and the given fairness setting.
     *
     * @param permits the initial number of permits available.
     *        This value may be negative, in which case releases
     *        must occur before any acquires will be granted.
     * @param fair {@code true} if this semaphore will guarantee
     *        first-in first-out granting of permits under contention,
     *        else {@code false}
     *     指定许可证个数，指定是否公平版本
     */
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

内部类

Semaphore 内部包含公平锁（FairSync）和非公平锁（NonfairSync），继承内部类 Sync ，其中 Sync 继承 AQS

Sync

    /**
     * Synchronization implementation for semaphore.  Uses AQS state
     * to represent permits. Subclassed into fair and nonfair
     * versions.
     * 信号量的同步实现。使用AQS状态表示许可证。子类分为公平和非公平版本。
     */
    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;

        Sync(int permits) {
            setState(permits);
        }

        final int getPermits() {
            return getState();
        }

        //非公平模式获取许可
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining  0 时，获取该资源并使 Semaphore – 1 。如果Semaphore 值 = 0，则表示全部的共享资源已经被其他线程全部占用，线程必须要等待其他线程释放资源。当线程释放资源时，Semaphore 则 +1 。 模拟停车场有3个停车位，有5辆车要去停车
 
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  0 || h == null || h.waitStatus