控制QPS
可使用 RateLimiter 来实现
RateLimiter 从概念上来讲，速率限制器会在可配置的速率下分配许可证。如果必要的话，每个acquire() 会阻塞当前线程直到许可证可用后获取该许可证。一旦获取到许可证，不需要再释放许可证。
RateLimiter使用的是一种叫令牌桶的流控算法，RateLimiter会按照一定的频率往桶里扔令牌，线程拿到令牌才能执行，比如你希望自己的应用程序QPS不要超过1000，那么RateLimiter设置1000的速率后，就会每秒往桶里扔1000个令牌。
RateLimiter经常用于限制对一些物理资源或者逻辑资源的访问速率。与Semaphore 相比，Semaphore 限制了并发访问的数量而不是使用速率。
通过设置许可证的速率来定义RateLimiter。在默认配置下，许可证会在固定的速率下被分配，速率单位是每秒多少个许可证。为了确保维护配置的速率，许可会被平稳地分配，许可之间的延迟会做调整。
可能存在配置一个拥有预热期的RateLimiter 的情况，在这段时间内，每秒分配的许可数会稳定地增长直到达到稳定的速率。
RateLimiter同时也支持动态调整速率，这个通过setRate方法实现
|
1 2 3 4 5 6 7 8 9 |
//速率是每秒两个许可 final RateLimiter rateLimiter = RateLimiter.create(2.0); void submitTasks(List tasks, Executor executor) { for (Runnable task : tasks) { rateLimiter.acquire(); // 也许需要等待 executor.execute(task); } } |
再举另外一个例子，想象下我们制造了一个数据流，并希望以每秒5kb的速率处理它。可以通过要求每个字节代表一个许可，然后指定每秒5000个许可来完成：
|
1 2 3 4 5 6 7 |
// 每秒5000个许可 final RateLimiter rateLimiter = RateLimiter.create(5000.0); void submitPacket(byte[] packet) { rateLimiter.acquire(packet.length); networkService.send(packet); } |
控制并发线程数
同一瞬间同时访问方法块的并发线程数，可以使用java 自带的Semaphore来实现
Semaphore（信号量）是用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源。很多年以来，我都觉得从字面上很难理解Semaphore所表达的含义，只能把它比作是控制流量的红绿灯，比如XX马路要限制流量，只允许同时有一百辆车在这条路上行使，其他的都必须在路口等待，所以前一百辆车会看到绿灯，可以开进这条马路，后面的车会看到红灯，不能驶入XX马路，但是如果前一百辆中有五辆车已经离开了XX马路，那么后面就允许有5辆车驶入马路，这个例子里说的车就是线程，驶入马路就表示线程在执行，离开马路就表示线程执行完成，看见红灯就表示线程被阻塞，不能执行。
Semaphore可以用于做流量控制，特别公用资源有限的应用场景，比如数据库连接。假如有一个需求，要读取几万个文件的数据，因为都是IO密集型任务，我们可以启动几十个线程并发的读取，但是如果读到内存后，还需要存储到数据库中，而数据库的连接数只有10个，这时我们必须控制只有十个线程同时获取数据库连接保存数据，否则会报错无法获取数据库连接。这个时候，我们就可以使用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 |
public class SemaphoreTest { private static final int THREAD_COUNT = 30; private static ExecutorService threadPool = Executors .newFixedThreadPool(THREAD_COUNT); private static Semaphore s = new Semaphore(10); public static void main(String[] args) { for (int i = 0; i < THREAD_COUNT; i++) { threadPool.execute(new Runnable() { @Override public void run() { try { s.acquire(); System.out.println("save data"); s.release(); } catch (InterruptedException e) { } } }); } threadPool.shutdown(); } } |
在代码中，虽然有30个线程在执行，但是只允许10个并发的执行。Semaphore的构造方法Semaphore(int permits) 接受一个整型的数字，表示可用的许可证数量。Semaphore(10)表示允许10个线程获取许可证，也就是最大并发数是10。Semaphore的用法也很简单，首先线程使用Semaphore的acquire()获取一个许可证，使用完之后调用release()归还许可证。还可以用tryAcquire()方法尝试获取许可证。
其他方法
Semaphore还提供一些其他方法：
- int availablePermits() ：返回此信号量中当前可用的许可证数。
- int getQueueLength()：返回正在等待获取许可证的线程数。
- boolean hasQueuedThreads() ：是否有线程正在等待获取许可证。
- void reducePermits(int reduction) ：减少reduction个许可证。是个protected方法。
- Collection getQueuedThreads() ：返回所有等待获取许可证的线程集合。是个protected方法。
动态控制并发线程数
使用方法和Semaphore基本一致，对外新增了setPermits来重新设置同一时间最大的许可。在使用过程中可以动态更改
|
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 |
import lombok.extern.slf4j.Slf4j; import java.util.concurrent.Semaphore; /** * 动态信号量 * @author by xmh on 2018/8/2. */ @Slf4j public class DynamicSemaphore { private CustomSemaphore semaphore=null; private int maxPermits; public DynamicSemaphore(int permits){ this.maxPermits=permits; semaphore=new CustomSemaphore(permits); } public void acquire() throws InterruptedException { semaphore.acquire(); } public void release() { semaphore.release(); } public synchronized void setPermits(int newMax){ if (newMax < 1) { throw new IllegalArgumentException("Semaphore size must be at least 1," + " was " + newMax); } int delta = newMax - maxPermits; if(delta!=0){ log.info("setPermits maxPermits:{} newMax:{}",maxPermits,newMax); } if(delta>0){ //释放多余的许可 semaphore.release(delta); }else { //减少许可 delta *= -1; semaphore.reducePermits(delta); } this.maxPermits=newMax; } /** * Semaphore whose available permits change according to the * changes in max-pool-size via a reconfiguration. */ private static final class CustomSemaphore extends Semaphore { CustomSemaphore(int permits) { super(permits); } @Override protected void reducePermits(int size) { super.reducePermits(size); } } } |