Semaphore的使用并实现的限流器

秒速五厘米 2022-12-03 02:24 278阅读 0赞

# semaphore 基本原理 #

semaphore 翻译成字面意思是信号量，Semaphore可以控制同时访问的线程个数，通过acquire（）获取一个许可，如果没有就等待，而release()释放一个许可。  
从功能看semaphore的底层是基于AQS实现的一个共享锁，需要多个现场共享一个令牌池。创建semaphore实例的时候，需要一个参数permits，这个基本可以确定是设置给AQS的state的，然后每一个线程调用的acquire的时候，执行state = state-1，release的时候执行state = state +1，当然，如果state = 0 了，说明没有资源了，需要等待其他线程release。

Semaphore类位于java.util.concurrent包下，提供了2个构造器

public Semaphore(int permits) { 
        // 参数permits表示许可数目，可以多少个线程同时访问
            sync = new NonfairSync(permits);
        }
        public Semaphore(int permits, boolean fair) {  
        // 这个多了一个参数fair表示是否是公平的，即等待时间越久的越先获取许可
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }

下面说一下Semaphore类中比较重要的几个方法，首先是acquire()、release()方法：

public void acquire() throws InterruptedException {   }     //获取一个许可
    public void acquire(int permits) throws InterruptedException {  }    //获取permits个许可
    public void release() {  }          //释放一个许可
    public void release(int permits) {  }    //释放permits个许可

acquire()用来获取一个许可，若无许可能够获得，则会一直等待，直到获得许可。  
release()用来释放许可。注意，在释放许可之前，必须先获获得许可。  
这4个方法都会被阻塞，如果想立即得到执行结果，可以使用下面几个方法：

public boolean tryAcquire() {  };    //尝试获取一个许可，若获取成功，则立即返回true，若获取失败，则立即返回false
    public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {  };  //尝试获取一个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false
    public boolean tryAcquire(int permits) {  }; //尝试获取permits个许可，若获取成功，则立即返回true，若获取失败，则立即返回false
    public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException {  }; //尝试获取permits个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false

下面通过一个例子来看一下Semaphore的具体使用：  
假若一个工厂有5台机器，但是有8个工人，一台机器同时只能被一个工人使用，只有使用完了，其他工人才能继续使用。那么我们就可以通过Semaphore来实现：

import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.concurrent.Semaphore;
    
    public class SemaphoreTest { 
    
        public static void main(String[] args){ 
    
            final SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss SSS");
    
            final Semaphore semaphore = new Semaphore(5);
    
            for(int i = 0;i<8;i++){ 
    
                // 定义8个线程
    
                new Thread("工人"+i){ 
                    @Override
                    public void run() { 
    
                        try { 
                            semaphore.acquire();
    
                            System.out.println(sdf.format(new Date()) + " " + Thread.currentThread().getName() +"获得一个许可，并开始执行任务");
                            Thread.sleep(3000);
                            System.out.println(sdf.format(new Date()) + " " + Thread.currentThread().getName() +"执行完毕，开始释放许可");
                            semaphore.release();
    
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                }.start();
            }
        }
    }

执行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70_pic_center]

## 实现一个简单的限流器 ##

import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.concurrent.*;
    
    
    public class SemaphoreLimitTest { 
    
        // 定义一个执行线程池
        private final Executor executor = new ThreadPoolExecutor(10, 20, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(10));
        // 每次只能执行5个任务
        final Semaphore semaphore = new Semaphore(5);
    
        public void process() { 
            executor.execute(new Runnable() { 
                @Override
                public void run() { 
                    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss SSS");
                    try { 
                        semaphore.acquire();
                        System.out.println(sdf.format(new Date()) + " " + Thread.currentThread().getName() + "获得一个许可，并开始执行任务");
                        Thread.sleep(3000);
                        System.out.println(sdf.format(new Date()) + " " + Thread.currentThread().getName() + "执行完毕，开始释放许可");
                        semaphore.release();
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            });
        }
    
    
        // 模拟测试
        public static void main(String[] args) { 
            final SemaphoreLimitTest semaphoreLimitTest = new SemaphoreLimitTest();
            // 同时进来8个任务
            for (int i = 0; i < 8; i++) { 
                // 定义8个线程
                new Thread("线程" + i) { 
                    @Override
                    public void run() { 
                        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss SSS");
                        System.out.println(sdf.format(new Date()) + " " + Thread.currentThread().getName() + "开始请求");
                        semaphoreLimitTest.process();
                    }
                }.start();
            }
        }
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70_pic_center 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/b14c16301e9644b2ab1e060eeb2a6da7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/915a87d233c748a29ec838b7ba84c40d.png