何时使用限流：
比如你希望自己的应用程序 QPS不要超过1000，那么RateLimiter设置1000的速率后，就会每秒往桶里 扔1000个令牌,RateLimiter经常用于限制对一些物理资源或者逻辑资源的访 问速率。
简介：
对于单机版的限流，可以使用Google 开源的 Guava项目，这个项目提供了Google在Java项目中使用一些核心库，包含集合(Collections)，缓存(Caching)，并发编程库(Concurrency)，常用注解(Common annotations)，String操作，I/O操作方面的众多非常实用的函数。
这个项目也包含了限流的功能，其原理是根据令牌桶算法来实现。
提供了两种限流策略：
● 平滑突发限流(SmoothBursty)
● 平滑预热限流(SmoothWarmingUp)实现。
依赖：
|
1 2 3 4 5 |
<dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>29.0-jre</version> </dependency> |
方法描述：
模拟场景（示例）：
场景一：
当我们希望某一个接口每秒的访问量不超过10次
|
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 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
package org.xhs.test;
import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;
import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.*;
/** * @Author: hu.chen * @Description: **/ public class Test {
/** * 存储接口名和令牌生成器的对应关系 */ private static Map<String, RateLimiter> interfaces = new ConcurrentHashMap<>();
/** * 线程池 */ private static ExecutorService threadPool = new ThreadPoolExecutor(10,15,3,TimeUnit.SECONDS,new ArrayBlockingQueue<>(100));
public static void main(String[] args) throws InterruptedException {
List<UserRequest> tasks = new ArrayList<UserRequest>(); // 准备工作，先初始化 10个线程（用户），这10个用户同时访问一个接口 for (int i = 1; i <= 12; i++) { String ip = "127.0.0." + i; String userName="chenhu_"+i; String interfaceName="user/find_"; tasks.add(new UserRequest(ip,userName,interfaceName)); }
// 先初始化好令牌生成器 for (UserRequest request : tasks) { // 根据接口名限流 RateLimiter rateLimiter = interfaces.get(request.getInterfaceName());
if(rateLimiter==null){ // 创建一个令牌生成器，每秒产生10个令牌 synchronized (interfaces) { if(rateLimiter==null) { rateLimiter = RateLimiter.create(10); // 将这个令牌生成器和具体的接口进行绑定 interfaces.put(request.getInterfaceName(),rateLimiter); } } } }
// 休眠一秒，让令牌生成器先生成令牌 Thread.sleep(1000);
for (UserRequest request : tasks) { // 根据接口名限流 RateLimiter rateLimiter = interfaces.get(request.getInterfaceName());
// 获取令牌桶中一个令牌，如果获取不到，则等待 timeout 时间，如果还获取不到，则返回false，反之则返回true // timeout设置为0，表示不等待 if(rateLimiter.tryAcquire(1,0,TimeUnit.SECONDS)){
// 得到令牌，处理请求 threadPool.execute(()->{ System.err.println("接口："+request.getInterfaceName()+" 访问还未达到上限，"+request.getUserName()+"可以访问"); }); }else { // 已经等待了10秒还获取不到令牌，进行其他业务处理 System.err.println("当前时间访问失败，"+request.getUserName()+"无法获取令牌"); } } }
private static class UserRequest { /** * 请求用户ip */ private String ip;
/** * 用户名 */ private String userName;
/** * 请求的接口名 */ private String interfaceName; public UserRequest(String ip, String userName, String interfaceName) { this.ip = ip; this.userName = userName; this.interfaceName = interfaceName; } public String getIp() {return ip;}
public String getUserName() { return userName;} public String getInterfaceName() {return interfaceName;} } } |
场景二：
当我们希望某一个用户或者ip，每秒的访问量不超过10
|
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 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
package org.xhs.test;
import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;
import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.*;
/** * @Author: hu.chen * @Description: **/ public class Test {
/** * 存储用户名和令牌生成器的对应关系 */ private static Map<String, RateLimiter> interfaces = new ConcurrentHashMap<>();
/** * 线程池 */ private static ExecutorService threadPool = new ThreadPoolExecutor(10,15,3,TimeUnit.SECONDS,new ArrayBlockingQueue<>(100));
public static void main(String[] args) throws InterruptedException {
List<UserRequest> tasks = new ArrayList<UserRequest>(); // 准备工作，先初始化 10个线程（用户），这10个用户同时访问一个接口 for (int i = 1; i <= 12; i++) { String ip = "127.0.0." + i; String userName="chenhu_"; String interfaceName="user/find_"+i; tasks.add(new UserRequest(ip,userName,interfaceName)); }
// 先初始化好令牌生成器 for (UserRequest request : tasks) { // 根据接口名限流 RateLimiter rateLimiter = interfaces.get(request.getUserName());
if(rateLimiter==null){ // 创建一个令牌生成器，每秒产生5个令牌 synchronized (interfaces) { if(rateLimiter==null) { rateLimiter = RateLimiter.create(10); // 将这个令牌生成器和具体的接口进行绑定 interfaces.put(request.getUserName(),rateLimiter); } } } }
// 休眠一秒，让令牌生成器先生成令牌 Thread.sleep(1000);
for (UserRequest request : tasks) { // 根据接口名限流 RateLimiter rateLimiter = interfaces.get(request.getUserName());
// 获取令牌桶中一个令牌，如果获取不到，则等待 timeout 时间，如果还获取不到，则返回false，反之则返回true // timeout设置为0，表示不等待 if(rateLimiter.tryAcquire(1,0,TimeUnit.SECONDS)){
// 得到令牌，处理请求 threadPool.execute(()->{ System.err.println("用户："+request.getUserName()+" 当前时间访问次数还未达到上限，可以访问"); }); }else { // 已经等待了10秒还获取不到令牌，进行其他业务处理 System.err.println("当前时间访问失败，"+request.getUserName()+"无法获取令牌"); } } }
private static class UserRequest { /** * 请求用户ip */ private String ip;
/** * 用户名 */ private String userName;
/** * 请求的接口名 */ private String interfaceName; public UserRequest(String ip, String userName, String interfaceName) { this.ip = ip; this.userName = userName; this.interfaceName = interfaceName; } public String getIp() {return ip;}
public String getUserName() { return userName;} public String getInterfaceName() {return interfaceName;} } } |
