本篇博客中的大部分内容并非原创,仅做整理作为学习资料使用
参考:https://www.cnblogs.com/tong-yuan/p/Semaphore.html
PROPAGATE 还是不太懂,大概就是因为并发,head的状态处于一个中间值,导致了无法向后传播了,所以需要增加一个状态 PROPAGATE。
问题
(1)Semaphore是什么?
(2)Semaphore具有哪些特性?
(3)Semaphore通常使用在什么场景中?
(4)Semaphore的许可次数是否可以动态增减?
(5)Semaphore如何实现限流?
简介
Semaphore,信号量,它保存了一系列的许可(permits),每次调用acquire()都将消耗一个许可,每次调用release()都将归还一个许可。
特性
Semaphore通常用于限制同一时间对共享资源的访问次数上,也就是常说的限流。
下面我们一起来学习Java中Semaphore是如何实现的。
类结构
Semaphore中包含了一个实现了AQS的同步器Sync,以及它的两个子类FairSync和NonFairSync,这说明Semaphore也是区分公平模式和非公平模式的。
源码分析
基于之前对于ReentrantLock和ReentrantReadWriteLock的分析,这篇文章相对来说比较简单,之前讲过的一些方法将直接略过,有兴趣的可以拉到文章底部查看之前的文章。
内部类Sync
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67
| 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; } } protected final boolean tryReleaseShared(int releases) { for (;;) { int current = getState(); int next = current + releases; if (next < current) throw new Error("Maximum permit count exceeded"); if (compareAndSetState(current, next)) return true; } } final void reducePermits(int reductions) { for (;;) { int current = getState(); int next = current - reductions; if (next > current) throw new Error("Permit count underflow"); if (compareAndSetState(current, next)) return; } } final int drainPermits() { for (;;) { int current = getState(); if (current == 0 || compareAndSetState(current, 0)) return current; } } }
|
通过Sync的几个实现方法,我们获取到以下几点信息:
(1)许可是在构造方法时传入的;
(2)许可存放在状态变量state中;
(3)尝试获取一个许可的时候,则state的值减1;
(4)当state的值为0的时候,则无法再获取许可;
(5)释放一个许可的时候,则state的值加1;
(6)许可的个数可以动态改变;
内部类NonfairSync
1 2 3 4 5 6 7 8 9 10 11 12
| static final class NonfairSync extends Sync { private static final long serialVersionUID = -2694183684443567898L; NonfairSync(int permits) { super(permits); } protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires); } }
|
非公平模式下,直接调用父类的nonfairTryAcquireShared()尝试获取许可。
内部类FairSync
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23
| static final class FairSync extends Sync { private static final long serialVersionUID = 2014338818796000944L; FairSync(int permits) { super(permits); } protected int tryAcquireShared(int acquires) { for (;;) { if (hasQueuedPredecessors()) return -1; int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } } }
|
公平模式下,先检测前面是否有排队的,如果有排队的则获取许可失败,进入队列排队,否则尝试原子更新state的值。
构造方法
1 2 3 4 5 6 7 8
| public Semaphore(int permits) { sync = new NonfairSync(permits); }
public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }
|
创建Semaphore时需要传入许可次数。
Semaphore默认也是非公平模式,但是你可以调用第二个构造方法声明其为公平模式。
下面的方法在学习过前面的内容看来都比较简单,彤哥这里只列举Semaphore支持的一些功能了。
以下的方法都是针对非公平模式来描述。
acquire()方法
1 2 3
| public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); }
|
获取一个许可,默认使用的是可中断方式,如果尝试获取许可失败,会进入AQS的队列中排队。
acquireUninterruptibly()方法
1 2 3
| public void acquireUninterruptibly() { sync.acquireShared(1); }
|
获取一个许可,非中断方式,如果尝试获取许可失败,会进入AQS的队列中排队。
tryAcquire()方法
1 2 3
| public boolean tryAcquire() { return sync.nonfairTryAcquireShared(1) >= 0; }
|
尝试获取一个许可,使用Sync的非公平模式尝试获取许可方法,不论是否获取到许可都返回,只尝试一次,不会进入队列排队。
tryAcquire(long timeout, TimeUnit unit)方法
1 2 3 4
| public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); }
|
尝试获取一个许可,先尝试一次获取许可,如果失败则会等待timeout时间,这段时间内都没有获取到许可,则返回false,否则返回true;
release()方法
1 2 3
| public void release() { sync.releaseShared(1); }
|
释放一个许可,释放一个许可时state的值会加1,并且会唤醒下一个等待获取许可的线程。
acquire(int permits)方法
1 2 3 4
| public void acquire(int permits) throws InterruptedException { if (permits < 0) throw new IllegalArgumentException(); sync.acquireSharedInterruptibly(permits); }
|
一次获取多个许可,可中断方式。
acquireUninterruptibly(int permits)方法
1 2 3 4
| public void acquireUninterruptibly(int permits) { if (permits < 0) throw new IllegalArgumentException(); sync.acquireShared(permits); }
|
一次获取多个许可,非中断方式。
tryAcquire(int permits)方法
1 2 3 4
| public boolean tryAcquire(int permits) { if (permits < 0) throw new IllegalArgumentException(); return sync.nonfairTryAcquireShared(permits) >= 0; }
|
一次尝试获取多个许可,只尝试一次。
tryAcquire(int permits, long timeout, TimeUnit unit)方法
1 2 3 4 5
| public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException { if (permits < 0) throw new IllegalArgumentException(); return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout)); }
|
尝试获取多个许可,并会等待timeout时间,这段时间没获取到许可则返回false,否则返回true。
release(int permits)方法
1 2 3 4
| public void release(int permits) { if (permits < 0) throw new IllegalArgumentException(); sync.releaseShared(permits); }
|
一次释放多个许可,state的值会相应增加permits的数量。
availablePermits()方法
1 2 3
| public int availablePermits() { return sync.getPermits(); }
|
获取可用的许可次数。
drainPermits()方法
1 2 3
| public int drainPermits() { return sync.drainPermits(); }
|
销毁当前可用的许可次数,对于已经获取的许可没有影响,会把当前剩余的许可全部销毁。
reducePermits(int reduction)方法
1 2 3 4
| protected void reducePermits(int reduction) { if (reduction < 0) throw new IllegalArgumentException(); sync.reducePermits(reduction); }
|
减少许可的次数。
总结
(1)Semaphore,也叫信号量,通常用于控制同一时刻对共享资源的访问上,也就是限流场景;
(2)Semaphore的内部实现是基于AQS的共享锁来实现的;
(3)Semaphore初始化的时候需要指定许可的次数,许可的次数是存储在state中;
(4)获取一个许可时,则state值减1;
(5)释放一个许可时,则state值加1;
(6)可以动态减少n个许可;
(7)可以动态增加n个许可吗?
彩蛋
(1)如何动态增加n个许可?
答:调用release(int permits)即可。我们知道释放许可的时候state的值会相应增加,再回头看看释放许可的源码,发现与ReentrantLock的释放锁还是有点区别的,Semaphore释放许可的时候并不会检查当前线程有没有获取过许可,所以可以调用释放许可的方法动态增加一些许可。
(2)如何实现限流?
答:限流,即在流量突然增大的时候,上层要能够限制住突然的大流量对下游服务的冲击,在分布式系统中限流一般做在网关层,当然在个别功能中也可以自己简单地来限流,比如秒杀场景,假如只有10个商品需要秒杀,那么,服务本身可以限制同时只进来100个请求,其它请求全部作废,这样服务的压力也不会太大。
使用Semaphore就可以直接针对这个功能来限流,以下是代码实现:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
| public class SemaphoreTest { public static final Semaphore SEMAPHORE = new Semaphore(100); public static final AtomicInteger failCount = new AtomicInteger(0); public static final AtomicInteger successCount = new AtomicInteger(0);
public static void main(String[] args) { for (int i = 0; i < 1000; i++) { new Thread(()->seckill()).start(); } }
public static boolean seckill() { if (!SEMAPHORE.tryAcquire()) { System.out.println("no permits, count="+failCount.incrementAndGet()); return false; }
try { Thread.sleep(2000); System.out.println("seckill success, count="+successCount.incrementAndGet()); } catch (InterruptedException e) { e.printStackTrace(); } finally { SEMAPHORE.release(); } return true; } }
|