单机限流 - 低入侵限流框架设计与实现

对于单机限流的一些内容，可以先查看 单机限流 - 限流算法及隔离策略

限流算法 - 令牌桶图例

Guava RateLimiter

Guava不仅仅在集合、缓存、异步回调等方面功能强大，而且还给我们封装好了限流的API!
Guava RateLimiter基于令牌桶算法，我们只需要告诉RateLimiter系统限制的QPS是多少，那么RateLimiter将以这个速度往桶里面放入令牌，然后请求的时候，通过tryAcquire()方法向RateLimiter获取许可（令牌)。

低入侵限流框架设计

基于Guava RateLimiter 实现

        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>27.1-jre</version>
        </dependency>

import com.google.common.util.concurrent.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;

/**
 * 使用guava RateLimiter实现限流
 */
@RestController
@RequestMapping("limiter")
public class MyLimiterController {
    private Logger log = LoggerFactory.getLogger(MyLimiterController.class);

    // 以1r/s往桶子中放入令牌
    private RateLimiter rateLimiter = RateLimiter.create(1.0);

    private DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    @GetMapping("index")
    public String indexLimiter() {
        LocalDateTime now = LocalDateTime.now();

        // 500 毫秒内没有获取到令牌，就直接放弃获取进行服务降级处理
        boolean tryAcquire = rateLimiter.tryAcquire(500, TimeUnit.MILLISECONDS);
        if (!tryAcquire) {
            log.info("Error -- 时间：{} ，获取令牌失败。", now.format(dateTimeFormatter));
            return "系统繁忙，请稍后再试。";
        }
        log.info("Success -- 时间：{} ，获取令牌成功。", now.format(dateTimeFormatter));
        return "success";
    }

}

使用注解来实现指定接口的限流

import java.lang.annotation.*;
import java.util.concurrent.TimeUnit;

/**
 * 自定义限流注解
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
@Documented
public @interface MyRateLimiter {
    //以固定数值往令牌桶添加令牌
    double permitsPerSecond();

    //获取令牌最大等待时间
    long timeout();

    // 单位(例:分钟/秒/毫秒) 默认:毫秒
    TimeUnit timeunit() default TimeUnit.MILLISECONDS;

    // 无法获取令牌返回提示信息 默认值可以自行修改
    String msg() default "系统繁忙,请稍后再试.";
}

package com.study.current.limiting.aop;

import com.google.common.collect.Maps;
import com.google.common.util.concurrent.RateLimiter;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.PrintWriter;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.Objects;

/**
 * 使用aop的环绕通知来拦截注解,使用了一个ConcurrentMap来保存每个请求对应的令牌桶,
 * key是没有url请求,防止出现每个请求都会新建一个令牌桶这么会达不到限流效果
 */
@Aspect
@Component
public class RateLimiterAspect {
    private Logger logger = LoggerFactory.getLogger(RateLimiterAspect.class);

    /**
     * 使用url做为key,存放令牌桶 防止每次重新创建令牌桶
     */
    private Map<String, RateLimiter> limitMap = Maps.newConcurrentMap();

    @Pointcut("@annotation(com.xxx.xxx.xxx.MyRateLimiter)")
    public void myRateLimiter() {
    }

    @Around("myRateLimiter()")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        // 获取request,response
        HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
        HttpServletResponse response = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getResponse();
        // 或者url(存在map集合的key)
        String url = request.getRequestURI();
        // 获取自定义注解
        MyRateLimiter rateLimiter = getAllenRateLimiter(joinPoint);
        if (rateLimiter != null) {
            RateLimiter limiter = null;
            // 判断map集合中是否有创建有创建好的令牌桶
            if (!limitMap.containsKey(url)) {
                // 创建令牌桶
                limiter = RateLimiter.create(rateLimiter.permitsPerSecond());
                limitMap.put(url, limiter);
                logger.info("<<=================  请求{},创建令牌桶,容量{} 成功!!!", url, rateLimiter.permitsPerSecond());
            }
            limiter = limitMap.get(url);
            // 获取令牌
            boolean acquire = limiter.tryAcquire(rateLimiter.timeout(), rateLimiter.timeunit());

            if (!acquire) {
                responseResult(response, 500, rateLimiter.msg());
                return null;
            }
        }
        return joinPoint.proceed();
    }

    /**
     * 获取注解对象
     *
     * @param joinPoint 对象
     * @return ten LogAnnotation
     */
    private MyRateLimiter getAllenRateLimiter(final JoinPoint joinPoint) {
        Method[] methods = joinPoint.getTarget().getClass().getDeclaredMethods();
        String name = joinPoint.getSignature().getName();
        if (!StringUtils.isEmpty(name)) {
            for (Method method : methods) {
                MyRateLimiter annotation = method.getAnnotation(MyRateLimiter.class);
                if (!Objects.isNull(annotation) && name.equals(method.getName())) {
                    return annotation;
                }
            }
        }
        return null;
    }

    /**
     * 自定义响应结果
     *
     * @param response 响应
     * @param code     响应码
     * @param message  响应信息
     */
    private void responseResult(HttpServletResponse response, Integer code, String message) {
        response.resetBuffer();
        response.setHeader("Access-Control-Allow-Origin", "*");
        response.setHeader("Access-Control-Allow-Credentials", "true");
        response.setContentType("application/json");
        response.setCharacterEncoding("UTF-8");
        PrintWriter writer = null;
        try {
            writer = response.getWriter();
            writer.println("{\"code\":" + code + " ,\"message\" :\"" + message + "\"}");
            response.flushBuffer();
        } catch (IOException e) {
            logger.error(" 输入响应出错 e = {}", e.getMessage(), e);
        } finally {
            if (writer != null) {
                writer.flush();
                writer.close();
            }
        }
    }
}

如果觉得有收获就点个赞吧，更多知识，请点击关注查看我的主页信息哦~