多线程控制并发数目工具类Semaphore

r囧r小猫 2024-03-24 15:28 120阅读 0赞

#### 文章目录 ####

*   *  前言
     *  Semaphore原理
     *  Semaphore源码解析
     *   *  内部继承AQS保证同步
         *  acquire获取许可
         *  release释放许可
     *  实战演示
     *  总结

### 前言 ###

在多线程编码过程中，我们会用到多线程来提升运行效率。比如我们的Executors创建线程池，程序尽可能的压榨CPU资源来提升我们程序吞吐量。但是过度的使用线程，也会将我们CPU资源榨干，从而让我们系统不能正常的提供服务。故今天我们引入JUC并发包下面的semaphore并发类，该类可以同时允许定量线程执行从而达到控制并发的目的。

### Semaphore原理 ###

Semaphore并发类提供了两个核心方法：acquire()方法和release()方法。acquire()方法表示获取一个许可，如果没有则等待，release()方法则是释放对应的许可。Semaphore维护了当前访问的个数，通过提供同步机制来控制同时访问的个数。

### Semaphore源码解析 ###

#### 内部继承AQS保证同步 ####

Semaphore 与AQS关系图：  
![在这里插入图片描述][cbd804fb850e46468ef274ec95b1fa63.png]

进入java.util.concurrent 包下 Semaphore 并发类查看源码：

//继承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 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

如上所示，首先Semaphore并发内部类继承AbstractQueuedSynchronizer 同步阻塞类，所以可以得出Semaphore是通过AQS机制来保证同步。AQS不明白的同学可以看我之前的博文，大概就是内部一个state状态，获取到资源 state 就加一，释放资源 state 就减一，当 state == 0 的时候表示阻塞队列中的其他线程可以获取该资源。

继续查看源码：

/**
     * 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构造方法默认提供非公平锁同步构造方法，也提供了一个用户自定义是否公平锁的构造方法。是否公平锁同步直接影响阻塞队列中的哪个线程可以获取资源，公平锁是按照阻塞顺序获取资源，非公平锁是多个线程争抢资源。当然构造方法必须传入并发许可的总数，这个总数直接影响后续我们可以同时获取多少个许可。

#### acquire获取许可 ####

查看Semaphore 获取许可的源码：

//是否可以获取许可
    public boolean tryAcquire() {
        return sync.nonfairTryAcquireShared(1) >= 0;
    }
    //是否可以获取许可，并提供超时获取机制
    public boolean tryAcquire(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }
    //是否可以获取多个许可，并提供超时获取机制
    public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
    }
    //提供一个无参获取许可方法，默认获取一个许可
    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    
    //不间断的获取许可，如果获取不到就阻塞
    public void acquireUninterruptibly() {
        sync.acquireShared(1);
    }
    //不间断的获取多个许可，如果获取不到就阻塞
    public void acquireUninterruptibly(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireShared(permits);
    }

如源码所示，我获取许可有很多的方法。有是否可以获取许可，超时是否可以获取许可。当然还有获取许可的方法，以及阻塞获取许可的多个方法，这些方法本质上都是调用父AQS中的获取资源许可的方法，同学们可以选择自己适用的方法进行调用。

#### release释放许可 ####

查看Semaphore 释放许可的源码：

//默认释放一个许可
    public void release() {
        sync.releaseShared(1);
    }
    //同时释放多个许可
    public void release(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.releaseShared(permits);
    }
    //调用父AQS的释放资源的方法
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

如源码所示，Semaphore提供了多个释放许可的方法，我们可以根据实际需要选择释放许可的方法。  
值得注意的是release() 方法调用了自己内部类Sync的释放资源方法，而Sync又是继承AQS阻塞队列并调用了父类doReleaseShared() 释放资源的方法。

### 实战演示 ###

上面博文我们介绍了Semaphore并发类是一个同步类，它继承了AQS阻塞队列。Semaphore主要提供了acquire() 获取许可与release() 释放许可的方法，通过这两个方法我们可以实现线程并发数目的功能。下面我们简单模拟一下限制并发数目。

1、Semaphore测试类  
测试类提供一个定量5的线程池，用countdownlatch 让主线程等待子线程执行完成，Semaphore保证每次只有3个线程执行。为了保证测试结果可以追溯性，我们在业务逻辑中让线程睡眠3s。

/**
     * Semaphore test
     * @author senfel
     * @date 2023/4/6 11:38 
     * @return
     */
    @SpringBootTest
    class ConcurrentApplicationTests {
        /**
         * 日期格式
         */
        private static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:mm:ss");
    
        @Test
        void testSemaphore() throws Exception {
            //创建线程为5的线程池
            ExecutorService executorService = Executors.newFixedThreadPool(5);
            //同时运行3个线程运行
            Semaphore semaphore = new Semaphore(3);
            //等待执行完成
            CountDownLatch countDownLatch = new CountDownLatch(10);
            for(int i = 0;i<10;i++) {
                executorService.execute(new Runnable() {
                    @Override
                    public void run() {
                        try{
                            //获取许可
                            semaphore.acquire();
                            //业务逻辑
                            executeFun();
                            //释放许可
                            semaphore.release();
                            countDownLatch.countDown();
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }
                });
            }
            countDownLatch.await();
            System.err.println("执行完成");
            executorService.shutdown();
        }
    
    
        /**
         * 执行业务方法
         * @author senfel
         * @date 2023/4/6 11:13
         * @return void
         */
        private void executeFun() {
            try {
                String startTime = formatter.format(LocalDateTime.now());
                Thread.sleep(3000);
                System.err.println("当前执行线程："+Thread.currentThread().getName()+",开始时间："+startTime+",结束时间"+formatter.format(LocalDateTime.now()));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
    }

2、执行测试方法  
执行testSemaphore() 方法，我们可以在控制台看到如下结果：

> 当前执行线程：pool-1-thread-1,开始时间：11:37:00,结束时间11:37:03  
> 当前执行线程：pool-1-thread-3,开始时间：11:37:00,结束时间11:37:03  
> 当前执行线程：pool-1-thread-2,开始时间：11:37:00,结束时间11:37:03  
> 当前执行线程：pool-1-thread-2,开始时间：11:37:03,结束时间11:37:06  
> 当前执行线程：pool-1-thread-1,开始时间：11:37:03,结束时间11:37:06  
> 当前执行线程：pool-1-thread-3,开始时间：11:37:03,结束时间11:37:06  
> 当前执行线程：pool-1-thread-1,开始时间：11:37:06,结束时间11:37:09  
> 当前执行线程：pool-1-thread-4,开始时间：11:37:06,结束时间11:37:09  
> 当前执行线程：pool-1-thread-2,开始时间：11:37:06,结束时间11:37:09  
> 当前执行线程：pool-1-thread-5,开始时间：11:37:09,结束时间11:37:12  
> 执行完成

根据执行结果我们可以发现每3s有3个线程执行并输出，得出结论Semaphore可以保证每次只能同时3个线程执行！！！！

### 总结 ###

多线程控制并发数目工具类Semaphore内部继承AQS抽象阻塞队列，并继承了其同步获取、释放资源的方法。Semaphore提供了acquire()获取许可、release()释放许可方法，底层当然调用的是AQS内部方法来满足同步以保证限制并发线程运行数目。

[cbd804fb850e46468ef274ec95b1fa63.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/ecf9e728855d4e238e0319139bb97b45.png