限流(rate limiting)
是指在一定时间内,对某些资源的访问次数进行限制,以避免资源被滥用或过度消耗。限流可以防止服务器崩溃、保证用户体验、提高系统可用性。
限流的方法有很多种,常见的有以下几种:
漏桶算法:
漏桶算法通过一个固定大小的漏桶来模拟流量,当流量进入漏桶时,会以恒定的速率从漏桶中流出。如果流量超过漏桶的容量,则会被丢弃。
令牌桶算法:
令牌桶算法通过一个固定大小的令牌桶来模拟流量,当流量进入令牌桶时,会从令牌桶中取出一个令牌。如果令牌桶中没有令牌,则会拒绝该流量。
滑动窗口算法:
滑动窗口算法通过一个固定大小的滑动窗口来模拟流量,当流量进入滑动窗口时,会统计窗口内流量的数量。如果窗口内流量的数量超过了一定的阈值,则会拒绝该流量。
限流可以应用在很多场景,例如:
防止服务器崩溃:当服务器的请求量过大时,可以通过限流来防止服务器崩溃。
保证用户体验:当用户请求某个资源的频率过高时,可以通过限流来降低用户的等待时间。
提高系统可用性:当系统的某些资源被滥用或过度消耗时,可以通过限流来提高系统的可用性。
限流是一个非常重要的技术,它可以帮助我们提高系统的稳定性和可用性。在实际开发中,我们可以根据不同的场景选择合适的限流算法。
我们定义的注解使用到技术:redis,redisson,AOP,自定义注解等
用到的部分依赖,这里没有指定版本,可根据市场上的版本进行配置
<!--redisson--> <dependency> <groupId>org.redisson</groupId> <artifactId>redisson-spring-boot-starter</artifactId> </dependency> <dependency> <groupId>com.baomidou</groupId> <artifactId>lock4j-redisson-spring-boot-starter</artifactId> </dependency> <!-- Spring框架基本的核心工具 --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context-support</artifactId> </dependency> <!-- SpringWeb模块 --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-web</artifactId> </dependency> <!-- 自定义验证注解 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-validation</artifactId> </dependency> <!-- aop --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-aop</artifactId> </dependency> <!--常用工具类 --> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> </dependency> <!-- servlet包 --> <dependency> <groupId>jakarta.servlet</groupId> <artifactId>jakarta.servlet-api</artifactId> </dependency> <!-- hutool 工具类 --> <dependency> <groupId>cn.hutool</groupId> <artifactId>hutool-core</artifactId> </dependency> <dependency> <groupId>cn.hutool</groupId> <artifactId>hutool-http</artifactId> </dependency> <dependency> <groupId>cn.hutool</groupId> <artifactId>hutool-extra</artifactId> </dependency> <!-- lombok 工具类 --> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency> <!-- 自动生成YML配置关联JSON文件 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-configuration-processor</artifactId> </dependency> <!-- 版本升级 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-properties-migrator</artifactId> <scope>runtime</scope> </dependency> <!-- 代码生产工具 --> <dependency> <groupId>io.github.linpeilie</groupId> <artifactId>mapstruct-plus-spring-boot-starter</artifactId> </dependency> <!-- 离线IP地址定位库 --> <dependency> <groupId>org.lionsoul</groupId> <artifactId>ip2region</artifactId> </dependency>
这里定义限流枚举类:LimitType
public enum LimitType { /** * 默认策略全局限流 */ DEFAULT, /** * 根据请求者IP进行限流 */ IP, /** * 实例限流(集群多后端实例) */ CLUSTER }
定义注解,在后续的代码中使用进行限流
import java.lang.annotation.*; /** * 限流注解 * * @author Lion Li */ @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @Documented public @interface RateLimiter { /** * 限流key,支持使用Spring el表达式来动态获取方法上的参数值 * 格式类似于 #code.id #{#code} */ String key() default ""; /** * 限流时间,单位秒 */ int time() default 60; /** * 限流次数 */ int count() default 100; /** * 限流类型 */ LimitType limitType() default LimitType.DEFAULT; /** * 提示消息 支持国际化 格式为 {code} */ String message() default "{rate.limiter.message}"; }
这里提供一下第 3 步需要的redis 工具类,可以根据自己的需求进行部分方法进行复制。
@NoArgsConstructor(access = AccessLevel.PRIVATE) @SuppressWarnings(value = {"unchecked", "rawtypes"}) public class RedisUtils { private static final RedissonClient CLIENT = SpringUtils.getBean(RedissonClient.class); /** * 限流 * * @param key 限流key * @param rateType 限流类型 * @param rate 速率 * @param rateInterval 速率间隔 * @return -1 表示失败 */ public static long rateLimiter(String key, RateType rateType, int rate, int rateInterval) { RRateLimiter rateLimiter = CLIENT.getRateLimiter(key); rateLimiter.trySetRate(rateType, rate, rateInterval, RateIntervalUnit.SECONDS); if (rateLimiter.tryAcquire()) { return rateLimiter.availablePermits(); } else { return -1L; } } /** * 获取客户端实例 */ public static RedissonClient getClient() { return CLIENT; } /** * 发布通道消息 * * @param channelKey 通道key * @param msg 发送数据 * @param consumer 自定义处理 */ public static <T> void publish(String channelKey, T msg, Consumer<T> consumer) { RTopic topic = CLIENT.getTopic(channelKey); topic.publish(msg); consumer.accept(msg); } public static <T> void publish(String channelKey, T msg) { RTopic topic = CLIENT.getTopic(channelKey); topic.publish(msg); } /** * 订阅通道接收消息 * * @param channelKey 通道key * @param clazz 消息类型 * @param consumer 自定义处理 */ public static <T> void subscribe(String channelKey, Class<T> clazz, Consumer<T> consumer) { RTopic topic = CLIENT.getTopic(channelKey); topic.addListener(clazz, (channel, msg) -> consumer.accept(msg)); } /** * 缓存基本的对象,Integer、String、实体类等 * * @param key 缓存的键值 * @param value 缓存的值 */ public static <T> void setCacheObject(final String key, final T value) { setCacheObject(key, value, false); } /** * 缓存基本的对象,保留当前对象 TTL 有效期 * * @param key 缓存的键值 * @param value 缓存的值 * @param isSaveTtl 是否保留TTL有效期(例如: set之前ttl剩余90 set之后还是为90) * @since Redis 6.X 以上使用 setAndKeepTTL 兼容 5.X 方案 */ public static <T> void setCacheObject(final String key, final T value, final boolean isSaveTtl) { RBucket<T> bucket = CLIENT.getBucket(key); if (isSaveTtl) { try { bucket.setAndKeepTTL(value); } catch (Exception e) { long timeToLive = bucket.remainTimeToLive(); setCacheObject(key, value, Duration.ofMillis(timeToLive)); } } else { bucket.set(value); } } /** * 缓存基本的对象,Integer、String、实体类等 * * @param key 缓存的键值 * @param value 缓存的值 * @param duration 时间 */ public static <T> void setCacheObject(final String key, final T value, final Duration duration) { RBatch batch = CLIENT.createBatch(); RBucketAsync<T> bucket = batch.getBucket(key); bucket.setAsync(value); bucket.expireAsync(duration); batch.execute(); } /** * 如果不存在则设置 并返回 true 如果存在则返回 false * * @param key 缓存的键值 * @param value 缓存的值 * @return set成功或失败 */ public static <T> boolean setObjectIfAbsent(final String key, final T value, final Duration duration) { RBucket<T> bucket = CLIENT.getBucket(key); return bucket.setIfAbsent(value, duration); } /** * 如果存在则设置 并返回 true 如果存在则返回 false * * @param key 缓存的键值 * @param value 缓存的值 * @return set成功或失败 */ public static <T> boolean setObjectIfExists(final String key, final T value, final Duration duration) { RBucket<T> bucket = CLIENT.getBucket(key); return bucket.setIfExists(value, duration); } /** * 注册对象监听器 * <p> * key 监听器需开启 `notify-keyspace-events` 等 redis 相关配置 * * @param key 缓存的键值 * @param listener 监听器配置 */ public static <T> void addObjectListener(final String key, final ObjectListener listener) { RBucket<T> result = CLIENT.getBucket(key); result.addListener(listener); } /** * 设置有效时间 * * @param key Redis键 * @param timeout 超时时间 * @return true=设置成功;false=设置失败 */ public static boolean expire(final String key, final long timeout) { return expire(key, Duration.ofSeconds(timeout)); } /** * 设置有效时间 * * @param key Redis键 * @param duration 超时时间 * @return true=设置成功;false=设置失败 */ public static boolean expire(final String key, final Duration duration) { RBucket rBucket = CLIENT.getBucket(key); return rBucket.expire(duration); } /** * 获得缓存的基本对象。 * * @param key 缓存键值 * @return 缓存键值对应的数据 */ public static <T> T getCacheObject(final String key) { RBucket<T> rBucket = CLIENT.getBucket(key); return rBucket.get(); } /** * 获得key剩余存活时间 * * @param key 缓存键值 * @return 剩余存活时间 */ public static <T> long getTimeToLive(final String key) { RBucket<T> rBucket = CLIENT.getBucket(key); return rBucket.remainTimeToLive(); } /** * 删除单个对象 * * @param key 缓存的键值 */ public static boolean deleteObject(final String key) { return CLIENT.getBucket(key).delete(); } /** * 删除集合对象 * * @param collection 多个对象 */ public static void deleteObject(final Collection collection) { RBatch batch = CLIENT.createBatch(); collection.forEach(t -> { batch.getBucket(t.toString()).deleteAsync(); }); batch.execute(); } /** * 检查缓存对象是否存在 * * @param key 缓存的键值 */ public static boolean isExistsObject(final String key) { return CLIENT.getBucket(key).isExists(); } /** * 缓存List数据 * * @param key 缓存的键值 * @param dataList 待缓存的List数据 * @return 缓存的对象 */ public static <T> boolean setCacheList(final String key, final List<T> dataList) { RList<T> rList = CLIENT.getList(key); return rList.addAll(dataList); } /** * 追加缓存List数据 * * @param key 缓存的键值 * @param data 待缓存的数据 * @return 缓存的对象 */ public static <T> boolean addCacheList(final String key, final T data) { RList<T> rList = CLIENT.getList(key); return rList.add(data); } /** * 注册List监听器 * <p> * key 监听器需开启 `notify-keyspace-events` 等 redis 相关配置 * * @param key 缓存的键值 * @param listener 监听器配置 */ public static <T> void addListListener(final String key, final ObjectListener listener) { RList<T> rList = CLIENT.getList(key); rList.addListener(listener); } /** * 获得缓存的list对象 * * @param key 缓存的键值 * @return 缓存键值对应的数据 */ public static <T> List<T> getCacheList(final String key) { RList<T> rList = CLIENT.getList(key); return rList.readAll(); } /** * 获得缓存的list对象(范围) * * @param key 缓存的键值 * @param form 起始下标 * @param to 截止下标 * @return 缓存键值对应的数据 */ public static <T> List<T> getCacheListRange(final String key, int form, int to) { RList<T> rList = CLIENT.getList(key); return rList.range(form, to); } /** * 缓存Set * * @param key 缓存键值 * @param dataSet 缓存的数据 * @return 缓存数据的对象 */ public static <T> boolean setCacheSet(final String key, final Set<T> dataSet) { RSet<T> rSet = CLIENT.getSet(key); return rSet.addAll(dataSet); } /** * 追加缓存Set数据 * * @param key 缓存的键值 * @param data 待缓存的数据 * @return 缓存的对象 */ public static <T> boolean addCacheSet(final String key, final T data) { RSet<T> rSet = CLIENT.getSet(key); return rSet.add(data); } /** * 注册Set监听器 * <p> * key 监听器需开启 `notify-keyspace-events` 等 redis 相关配置 * * @param key 缓存的键值 * @param listener 监听器配置 */ public static <T> void addSetListener(final String key, final ObjectListener listener) { RSet<T> rSet = CLIENT.getSet(key); rSet.addListener(listener); } /** * 获得缓存的set * * @param key 缓存的key * @return set对象 */ public static <T> Set<T> getCacheSet(final String key) { RSet<T> rSet = CLIENT.getSet(key); return rSet.readAll(); } /** * 缓存Map * * @param key 缓存的键值 * @param dataMap 缓存的数据 */ public static <T> void setCacheMap(final String key, final Map<String, T> dataMap) { if (dataMap != null) { RMap<String, T> rMap = CLIENT.getMap(key); rMap.putAll(dataMap); } } /** * 注册Map监听器 * <p> * key 监听器需开启 `notify-keyspace-events` 等 redis 相关配置 * * @param key 缓存的键值 * @param listener 监听器配置 */ public static <T> void addMapListener(final String key, final ObjectListener listener) { RMap<String, T> rMap = CLIENT.getMap(key); rMap.addListener(listener); } /** * 获得缓存的Map * * @param key 缓存的键值 * @return map对象 */ public static <T> Map<String, T> getCacheMap(final String key) { RMap<String, T> rMap = CLIENT.getMap(key); return rMap.getAll(rMap.keySet()); } /** * 获得缓存Map的key列表 * * @param key 缓存的键值 * @return key列表 */ public static <T> Set<String> getCacheMapKeySet(final String key) { RMap<String, T> rMap = CLIENT.getMap(key); return rMap.keySet(); } /** * 往Hash中存入数据 * * @param key Redis键 * @param hKey Hash键 * @param value 值 */ public static <T> void setCacheMapValue(final String key, final String hKey, final T value) { RMap<String, T> rMap = CLIENT.getMap(key); rMap.put(hKey, value); } /** * 获取Hash中的数据 * * @param key Redis键 * @param hKey Hash键 * @return Hash中的对象 */ public static <T> T getCacheMapValue(final String key, final String hKey) { RMap<String, T> rMap = CLIENT.getMap(key); return rMap.get(hKey); } /** * 删除Hash中的数据 * * @param key Redis键 * @param hKey Hash键 * @return Hash中的对象 */ public static <T> T delCacheMapValue(final String key, final String hKey) { RMap<String, T> rMap = CLIENT.getMap(key); return rMap.remove(hKey); } /** * 删除Hash中的数据 * * @param key Redis键 * @param hKeys Hash键 */ public static <T> void delMultiCacheMapValue(final String key, final Set<String> hKeys) { RBatch batch = CLIENT.createBatch(); RMapAsync<String, T> rMap = batch.getMap(key); for (String hKey : hKeys) { rMap.removeAsync(hKey); } batch.execute(); } /** * 获取多个Hash中的数据 * * @param key Redis键 * @param hKeys Hash键集合 * @return Hash对象集合 */ public static <K, V> Map<K, V> getMultiCacheMapValue(final String key, final Set<K> hKeys) { RMap<K, V> rMap = CLIENT.getMap(key); return rMap.getAll(hKeys); } /** * 设置原子值 * * @param key Redis键 * @param value 值 */ public static void setAtomicValue(String key, long value) { RAtomicLong atomic = CLIENT.getAtomicLong(key); atomic.set(value); } /** * 获取原子值 * * @param key Redis键 * @return 当前值 */ public static long getAtomicValue(String key) { RAtomicLong atomic = CLIENT.getAtomicLong(key); return atomic.get(); } /** * 递增原子值 * * @param key Redis键 * @return 当前值 */ public static long incrAtomicValue(String key) { RAtomicLong atomic = CLIENT.getAtomicLong(key); return atomic.incrementAndGet(); } /** * 递减原子值 * * @param key Redis键 * @return 当前值 */ public static long decrAtomicValue(String key) { RAtomicLong atomic = CLIENT.getAtomicLong(key); return atomic.decrementAndGet(); } /** * 获得缓存的基本对象列表 * * @param pattern 字符串前缀 * @return 对象列表 */ public static Collection<String> keys(final String pattern) { Stream<String> stream = CLIENT.getKeys().getKeysStreamByPattern(pattern); return stream.collect(Collectors.toList()); } /** * 删除缓存的基本对象列表 * * @param pattern 字符串前缀 */ public static void deleteKeys(final String pattern) { CLIENT.getKeys().deleteByPattern(pattern); } /** * 检查redis中是否存在key * * @param key 键 */ public static Boolean hasKey(String key) { RKeys rKeys = CLIENT.getKeys(); return rKeys.countExists(key) > 0; } }
提供一下第 3 步需要 获取i18n资源文件 类,可以做国际化进行处理,如果项目没有国际化,这个可以省略
@NoArgsConstructor(access = AccessLevel.PRIVATE) public class MessageUtils { private static final MessageSource MESSAGE_SOURCE = SpringUtils.getBean(MessageSource.class); /** * 根据消息键和参数 获取消息 委托给spring messageSource * * @param code 消息键 * @param args 参数 * @return 获取国际化翻译值 */ public static String message(String code, Object... args) { try { return MESSAGE_SOURCE.getMessage(code, args, LocaleContextHolder.getLocale()); } catch (NoSuchMessageException e) { return code; } } }
这个我们再自定义一个业务异常类,用于抛出异常 ,如果自己项目之前有定义,也可以使用自己的异常类
ServiceException
@Data @EqualsAndHashCode(callSuper = true) @NoArgsConstructor @AllArgsConstructor public final class ServiceException extends RuntimeException { @Serial private static final long serialVersionUID = 1L; /** * 错误码 */ private Integer code; /** * 错误提示 */ private String message; /** * 错误明细,内部调试错误 */ private String detailMessage; public ServiceException(String message) { this.message = message; } public ServiceException(String message, Integer code) { this.message = message; this.code = code; } public String getDetailMessage() { return detailMessage; } @Override public String getMessage() { return message; } public Integer getCode() { return code; } public ServiceException setMessage(String message) { this.message = message; return this; } public ServiceException setDetailMessage(String detailMessage) { this.detailMessage = detailMessage; return this; } }
如果对 ip 进行限流,在注解处理中会用到参数,ip ,url 等信息
ServletUtils
@NoArgsConstructor(access = AccessLevel.PRIVATE) public class ServletUtils extends JakartaServletUtil { /** * 获取request */ public static HttpServletRequest getRequest() { try { return getRequestAttributes().getRequest(); } catch (Exception e) { return null; } } public static String getClientIP() { return getClientIP(getRequest()); } }
处理限流注解:RateLimiterAspect
对注解处理的核心代码就在这里,
@Slf4j @Aspect public class RateLimiterAspect { /** * 定义spel表达式解析器 */ private final ExpressionParser parser = new SpelExpressionParser(); /** * 定义spel解析模版 */ private final ParserContext parserContext = new TemplateParserContext(); /** * 定义spel上下文对象进行解析 */ private final EvaluationContext context = new StandardEvaluationContext(); /** * 方法参数解析器 */ private final ParameterNameDiscoverer pnd = new DefaultParameterNameDiscoverer(); /** * GLOBAL_REDIS_KEY 和 RATE_LIMIT_KEY 最好还是定义在项目的一个统一的常量文件中,这里为了解剖出来的文件少一点 * * */ /** * 全局 redis key (业务无关的key) */ private final String GLOBAL_REDIS_KEY = "global:"; /** * 限流 redis key */ private final String RATE_LIMIT_KEY = GLOBAL_REDIS_KEY + "rate_limit:"; @Before("@annotation(rateLimiter)") public void doBefore(JoinPoint point, RateLimiter rateLimiter) throws Throwable { // 获取注解传的 时间 次数 int time = rateLimiter.time(); int count = rateLimiter.count(); // 处理 key String combineKey = getCombineKey(rateLimiter, point); try { RateType rateType = RateType.OVERALL; if (rateLimiter.limitType() == LimitType.CLUSTER) { rateType = RateType.PER_CLIENT; } long number = RedisUtils.rateLimiter(combineKey, rateType, count, time); if (number == -1) { String message = rateLimiter.message(); if (StringUtils.startsWith(message, "{") && StringUtils.endsWith(message, "}")) { message = MessageUtils.message(StringUtils.substring(message, 1, message.length() - 1)); } throw new ServiceException(message); } log.info("限制令牌 => {}, 剩余令牌 => {}, 缓存key => '{}'", count, number, combineKey); } catch (Exception e) { if (e instanceof ServiceException) { throw e; } else { throw new RuntimeException("服务器限流异常,请稍候再试"); } } } /** * 返回带有特定前缀的 key * @param rateLimiter 限流注解 * @param point 切入点 * @return key */ public String getCombineKey(RateLimiter rateLimiter, JoinPoint point) { String key = rateLimiter.key(); // 获取方法(通过方法签名来获取) MethodSignature signature = (MethodSignature) point.getSignature(); Method method = signature.getMethod(); Class<?> targetClass = method.getDeclaringClass(); // 判断是否是spel格式 if (StringUtils.containsAny(key, "#")) { // 获取参数值 Object[] args = point.getArgs(); // 获取方法上参数的名称 String[] parameterNames = pnd.getParameterNames(method); if (ArrayUtil.isEmpty(parameterNames)) { throw new ServiceException("限流key解析异常!请联系管理员!"); } for (int i = 0; i < parameterNames.length; i++) { context.setVariable(parameterNames[i], args[i]); } // 解析返回给key try { Expression expression; if (StringUtils.startsWith(key, parserContext.getExpressionPrefix()) && StringUtils.endsWith(key, parserContext.getExpressionSuffix())) { expression = parser.parseExpression(key, parserContext); } else { expression = parser.parseExpression(key); } key = expression.getValue(context, String.class) + ":"; } catch (Exception e) { throw new ServiceException("限流key解析异常!请联系管理员!"); } } // 限流前缀key StringBuilder stringBuffer = new StringBuilder(RATE_LIMIT_KEY); stringBuffer.append(ServletUtils.getRequest().getRequestURI()).append(":"); // 判断限流类型 if (rateLimiter.limitType() == LimitType.IP) { // 获取请求ip stringBuffer.append(ServletUtils.getClientIP()).append(":"); } else if (rateLimiter.limitType() == LimitType.CLUSTER) { // 获取客户端实例id stringBuffer.append(RedisUtils.getClient().getId()).append(":"); } return stringBuffer.append(key).toString(); } }
到这里注解就定义好了,接下来就可以进行测试和使用!!!
定义一个 Controller 来测试限流,这里返回的 R ,可以根据自己项目统一定义的返回,或者使用 void
RedisRateLimiterController
@Slf4j @RestController @RequestMapping("/demo/rateLimiter") public class RedisRateLimiterController { /** * 测试全局限流 * 全局影响 */ @RateLimiter(count = 2, time = 10) @GetMapping("/test") public R<String> test(String value) { return R.ok("操作成功", value); } /** * 测试请求IP限流 * 同一IP请求受影响 */ @RateLimiter(count = 2, time = 10, limitType = LimitType.IP) @GetMapping("/testip") public R<String> testip(String value) { return R.ok("操作成功", value); } /** * 测试集群实例限流 * 启动两个后端服务互不影响 */ @RateLimiter(count = 2, time = 10, limitType = LimitType.CLUSTER) @GetMapping("/testcluster") public R<String> testcluster(String value) { return R.ok("操作成功", value); } /** * 测试请求IP限流(key基于参数获取) * 同一IP请求受影响 * * 简单变量获取 #变量 复杂表达式 #{#变量 != 1 ? 1 : 0} */ @RateLimiter(count = 2, time = 10, limitType = LimitType.IP, key = "#value") @GetMapping("/testObj") public R<String> testObj(String value) { return R.ok("操作成功", value); } }
如果代码写的有问题,欢迎大家评论交流,进行指点!!!
也希望大家点个关注哦~~~~~~~~