Semaphore 掌握多线程限流的“秘钥”【Java多线程必备】

Dear 丶 2024-03-22 10:12 39阅读 0赞

## 一、介绍 ##

Semaphore（信号量）是Java中一个用于多线程编程的工具类，它提供了一种控制访问共享资源的机制，以避免多个线程同时访问同一个共享资源造成的冲突问题。Semaphore可以看做是一种计数器，用来控制同时访问某个共享资源的线程个数，也就是说，它可以控制并发线程的数量。

## 二、特性 ##

1. 计数器：Semaphore内部维护了一个计数器，表示当前可用的许可证数量。

2. 许可证：Semaphore可以控制同时访问某个共享资源的线程个数，每个线程需要获取一个许可证才能访问共享资源。当许可证数量为0时，其他线程需要等待。

3. 公平性：Semaphore可以选择公平模式或非公平模式。在公平模式下，等待时间最长的线程会优先获取许可证。

4. 释放许可证：当一个线程访问共享资源结束后，需要释放许可证以便其他线程可以访问。

5. 动态调整：Semaphore的许可证数量可以动态调整。

## 三、实现原理 ##

Semaphore的实现基于AQS（AbstractQueuedSynchronizer）类，AQS是Java中一个同步工具类的基础框架，Semaphore就是基于AQS的共享模式实现的。

Semaphore维护了一个内部的计数器，每次请求一个许可证时，计数器减一，释放一个许可证时，计数器加一。当计数器为0时，其他请求许可证的线程会被阻塞，直到有许可证可用。Semaphore内部维护了一个等待队列，当许可证不可用时，请求许可证的线程会被加入等待队列中。

Semaphore实现了公平性和非公平性两种模式。在公平模式下，等待时间最长的线程会优先获取许可证。在非公平模式下，请求许可证的线程会直接尝试获取许可证，不会排队等待。

## 四、应用场景 ##

1. 限流：在高并发的场景下，可以使用Semaphore来限制同时访问某个资源的线程数量，从而达到限流的目的。

2. 线程池：可以使用Semaphore来控制线程池中的线程数量，当线程池中的线程数量达到Semaphore的许可证数量时，新的任务会被阻塞等待。

3. 生产者消费者模式：可以使用Semaphore来控制生产者和消费者的数量，当生产者或消费者数量达到Semaphore的许可证数量时，另一个角色会被阻塞等待，从而实现生产者消费者模式的协调和控制。

4. 线程间通信：Semaphore可以用于线程间的通信，当线程需要等待某些条件满足时，可以使用Semaphore来进行阻塞等待，并在条件满足时释放许可证。

5. 限制资源并发访问：Semaphore可以用于限制某个资源的并发访问数量，从而避免因过多的并发访问而导致资源崩溃或性能下降等问题。

## 五、注意事项 ##

1. 许可证数量的设置：许可证数量需要根据实际场景进行设置，如果许可证数量过少，可能会导致线程一直被阻塞等待，从而影响系统性能；如果许可证数量过多，可能会导致系统资源的浪费。

2. 许可证的获取和释放：在使用Semaphore时，需要确保许可证的获取和释放是成对出现的，即每个线程获取到的许可证一定要及时释放，否则可能会导致其他线程一直被阻塞等待，从而影响系统性能。

3. 错误处理：在使用Semaphore时，需要考虑可能出现的异常情况，如许可证数量不足、线程被中断等，需要进行合理的错误处理，以保证程序的健壮性和可靠性。

4. 公平模式和非公平模式：需要根据实际场景选择使用公平模式还是非公平模式，如果要保证所有线程获取许可证的顺序是按照请求的顺序，那么需要使用公平模式；否则，可以选择非公平模式以提高系统的性能。

5. 多个Semaphore实例的协调和控制：如果有多个Semaphore实例需要进行协调和控制，需要根据实际场景进行合理的设计和组织，以保证系统的可靠性和高效性。

使用时需要注意合理设置许可证数量，确保许可证的获取和释放是成对出现的，并进行合理的错误处理和模式选择，以达到最佳的效果。

## 六、实际应用 ##

**1. 案例一**

**(1) 场景**

下面是一个使用Semaphore实现限流的示例代码。

**(2) 代码**

import java.util.concurrent.Semaphore;
    
    public class Resource {
        private final Semaphore semaphore;
    
        public Resource(int limit) {
            semaphore = new Semaphore(limit);
        }
    
        public void access() throws InterruptedException {
            semaphore.acquire();
            try {
                // do something
            } finally {
                semaphore.release();
            }
        }
    }

在这个示例代码中，我们创建了一个Resource类，其中包含了一个Semaphore对象。Resource类的access方法是需要被限流控制的方法，在方法开始时需要调用Semaphore的acquire方法获取许可证，如果许可证数量为0，则线程会被阻塞等待；在方法结束时需要调用Semaphore的release方法释放许可证。

> 通过这种方式，我们可以控制同时访问access方法的线程数量，从而实现限流的效果。除了上述内容，需要注意的是，Semaphore有两种模式：公平模式和非公平模式。在公平模式下，Semaphore会按照线程请求许可证的顺序分配许可证；而在非公平模式下，Semaphore会随机分配许可证，可能会导致某些线程一直无法获取许可证。默认情况下，Semaphore是非公平模式。

**2. 案例二**

**(1) 场景**

以下是一个使用Semaphore实现线程池的经典案例代码（注意：这仅是一个演示案例，帮助你快速理解。如果你想创建和使用线程池，请看这篇文章）。

**(2) 代码**

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 使用Semaphore实现线程池的经典案例
     *
     * @author wxy
     * @since 2023-04-23
     */
    public class SemaphoreThreadPool {
        private static final Logger LOGGER = LoggerFactory.getLogger(SemaphoreThreadPool.class);
    
        private final ExecutorService executor;
    
        private final Semaphore semaphore;
    
        /**
         * 构造
         *
         * @param poolSize 核心线程数&最大线程数
         */
        public SemaphoreThreadPool(int poolSize) {
            this.executor = Executors.newFixedThreadPool(poolSize);
            this.semaphore = new Semaphore(poolSize);
        }
    
        public void execute(Runnable task) throws InterruptedException {
            semaphore.acquire();
            executor.execute(() -> {
                try {
                    task.run();
                } finally {
                    semaphore.release();
                }
            });
        }
    
        public void shutdown() {
            executor.shutdown();
        }
    
        public static void main(String[] args) throws InterruptedException {
            // 创建实例并指定一次仅能执行3个线程
            SemaphoreThreadPool threadPool = new SemaphoreThreadPool(3);
            for (int i = 0; i < 10; i++) {
                threadPool.execute(() -> {
                    LOGGER.info("Thread {} is running.", Thread.currentThread().getName());
                    try {
                        TimeUnit.MILLISECONDS.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    LOGGER.info("Thread {} is finished.", Thread.currentThread().getName());
                });
            }
            threadPool.shutdown();
        }
    }

在上述代码中，Semaphore用于限制线程池中的线程数量，当线程池中的所有线程都正在执行任务时，新的任务会被阻塞等待。每当有一个线程完成任务时，会释放一个许可证，这样可以让其他被阻塞的任务继续执行。通过Semaphore的控制，可以有效地控制线程池的并发度（限制仅能执行你指定的或小于你指定的线程数量），从而保证系统的稳定性和性能。输出如下结果：

![d277d79c300e4eb2aa4cb85b9e798321.png][]

[d277d79c300e4eb2aa4cb85b9e798321.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/408057a02db84893951aa22905529633.png