Sentinel 限流

官方文档

https://github.com/alibaba/Sentinel/wiki/%E4%BB%8B%E7%BB%8D

Feign 中 使用 Sentinel 断路器

https://www.jianshu.com/p/8e769e365f88

在 SpringClouldGateway 的过滤器中进行QPS限流

引入依赖

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
</dependency>

添加限流的过滤器

LimitFilter

import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.csp.sentinel.slots.block.RuleConstant;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
import com.gnl.commonlib.error.SystemError;
import com.gnl.gateway.util.HttpDataExtractor;
import lombok.extern.log4j.Log4j2;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;


@Log4j2
@Component
public class LimitFilter implements GlobalFilter, Ordered {

    @Override
    public int getOrder() {
        // 返回值用来指定执行的顺序，数字越小，优先级越高
        return 1;
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String method = HttpDataExtractor.getMethod(exchange);
        String path = HttpDataExtractor.getPath(exchange);
        String userId = HttpDataExtractor.getHeader(exchange, "userId");

        // 被限流的资源名称（例如：GET /auth-service/auth/userInfo 123456）
        String resourceName = method + " " + path + " " + userId;

        // 不需要登录的接口，直接放行
        if (userId == null) {
            log.info(resourceName + " pass");
            return chain.filter(exchange);
        }

        // 初始化限流规则
        initFlowRules(resourceName);
        Entry entry = null;
        try {
            // 设置被限流的资源名称
            entry = SphU.entry(resourceName);
        } catch (BlockException e) {
            // 被限流时捕获BlockException,直接返回系统繁忙
            log.info(resourceName + " block");
            return HttpDataExtractor.returnData(SystemError.SYSTEM_IS_BUSY, exchange);
        } finally {
            if (entry != null) {
                entry.exit();
            }
        }

        // 未被限流，直接访问资源
        log.info(resourceName + " pass");
        return chain.filter(exchange);
    }

    private static void initFlowRules(String resourceName) {
        List<FlowRule> rules = new ArrayList<>();
        FlowRule rule = new FlowRule();
        rule.setResource(resourceName);
        rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        // 设置限流的 QPS 为 3，结合resourceName（GET /auth-service/auth/userInfo 666666），表示为用户（666666）访问接口GET /auth-service/auth/userInfo的QPS大于3时，就会被限流
        rule.setCount(3);
        rules.add(rule);
        FlowRuleManager.loadRules(rules);
    }

}

HttpDataExtractor

import com.alibaba.fastjson.JSON;
import com.gnl.commonlib.base.BaseError;
import com.gnl.commonlib.base.BaseResponse;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.function.Consumer;

public class HttpDataExtractor {

    /**
     * 获取请求头
     */
    public static String getHeader(ServerWebExchange exchange, String key) {
        ServerHttpRequest request = exchange.getRequest();
        List<String> list = request.getHeaders().get(key);
        if (list == null || list.size() == 0) {
            return null;
        }
        return list.get(0);
    }

    /**
     * 获得访问路径
     */
    public static String getPath(ServerWebExchange exchange) {
        ServerHttpRequest request = exchange.getRequest();
        return request.getURI().getPath();
    }

    /**
     * 获取请求方法
     */
    public static String getMethod(ServerWebExchange exchange) {
        ServerHttpRequest request = exchange.getRequest();
        return request.getMethod().toString();
    }

    /**
     * 添加请求头
     */
    public static void addHeader(ServerWebExchange exchange, Consumer<HttpHeaders> httpHeaders) {
        ServerHttpRequest request = exchange.getRequest();
        ServerHttpRequest serverHttpRequest = request.mutate().headers(httpHeaders).build();
        exchange.mutate().request(serverHttpRequest).build();
    }

    /**
     * 返回值的封装
     */
    public static Mono<Void> returnData(BaseError baseError, ServerWebExchange exchange) {
        ServerHttpResponse response = exchange.getResponse();
        BaseResponse data = BaseResponse.error(baseError);
        byte[] datas = JSON.toJSONString(data).getBytes(StandardCharsets.UTF_8);
        DataBuffer buffer = response.bufferFactory().wrap(datas);

        response.setStatusCode(HttpStatus.OK);
        response.getHeaders().add("Content-Type", "application/json;charset=UTF-8");

        return response.writeWith(Mono.just(buffer));
    }
}

效果

使用jMeter进行压测，0秒发送10个请求，只有三个成功访问了资源，后面七个直接返回了系统繁忙

image

成功返回的信息

image
被限流返回的信息
image