[大数据]网关限流及高可用_springCloud_6

一.基于漏桶算法的内置filter的限流

1.1 准备工作：

1.1.1 本地解压一个redis安装包，启动并监控

先打开服务：redis-server.exe，再打开客户端窗口：redis-cli.exe，打开客户端后输入： monitor命令进行监控
如下图所示：

1.1.2 工程中导入依赖：

在工程中引入redis相应的依赖：基于reactive的redis依赖

       <!--监控依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <!--redis的依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
        </dependency>

1.2 修改网关中的yml文件

server:
  port: 8080 #端口
spring:
  application:
    name: api-gateway-server #服务名称
  redis:
    host: localhost
    pool: 6379
    database: 0
  cloud: #配置SpringCloudGateway的路由
    gateway:
      routes:
      - id: product-service
        uri: lb://service-product
        predicates:
        - Path=/product-service/**
        filters:
        - name: RequestRateLimiter
          args:
            # 使用SpEL从容器中获取对象
            key-resolver: '#{@pathKeyResolver}'  #pathKeyResolver与配置类中的方法名一致
            # 令牌桶每秒填充平均速率
            redis-rate-limiter.replenishRate: 1
            # 令牌桶的上限
            redis-rate-limiter.burstCapacity: 3
        - RewritePath=/product-service/(?<segment>.*), /$\{segment}
      #RequestRateLimiter：使用限流过滤器，是sppringcloud gateway提供的
      #参数 replenishRate:向令牌桶内填充的速率
      #     burstCapacity:令牌桶的容量
#eureka注册中心
eureka:
  client:
    service-url:
      defaultZone: http://localhost:9000/eureka/
  instance:
    prefer-ip-address: true #使用ip地址注册

1.3 配置redis中key的解析器KeyResolve

package cn.itcast.gateway;

import org.springframework.cloud.gateway.filter.ratelimit.KeyResolver;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

/**
 * 限流规则配置类-存在多个KeyResolver会抛出异常
 * 1.基于路径的限流规则 ：abc
 * 2.基于请求ip的限流规则 127.0.0.1
 * 3.基于参数的
 */
@Configuration
public class KeyResolverConfiguration {

    /**
     * 编写基于请求路径的限流规则
     * 方法名称与yml配置文件中的key-resolver属性的@后面的名称一致
     */
    //@Bean
    public KeyResolver pathKeyResolver(){
        return new KeyResolver() {
            /**
             *
             * ServerWebExchange 上下文参数
             *
             */
            public Mono<String> resolve(ServerWebExchange exchange) {
                return  Mono.just(exchange.getRequest().getPath().toString());

            }
        };
    }

    /**
     * 基于lambad表达式的形式基于参数的限流
     *  格式 abc?access-token=1
     */
    @Bean
    public KeyResolver userKeyResolver(){
        return exchange -> Mono.just(
            //基于请求参数的限流
            exchange.getRequest().getQueryParams().getFirst("access-token")
            //基于ip地址的限流
            //  exchange.getRequest().getHeaders().getFirst("X-Forwarded-For")
        );
    }
}

二.基于Sentinel的限流

1.引入依赖：

<dependency>
      <groupId>com.alibaba.csp</groupId>
      <artifactId>sentinel-spring-cloud-gateway-adapter</artifactId>
      <version>1.6.3</version>
</dependency>

2.修改yml配置文件

注释基于filter的限流规则yml配置

server:
  port: 8080 #端口
spring:
  application:
    name: api-gateway-server #服务名称
  redis:
    host: localhost
    pool: 6379
    database: 0
  cloud: #配置SpringCloudGateway的路由
    gateway:
      routes:
      - id: product-service
        uri: lb://service-product
        predicates:
        - Path=/product-service/**
        filters:
#        - name: RequestRateLimiter
#          args:
#            # 使用SpEL从容器中获取对象
#            #key-resolver: '#{@pathKeyResolver}'  #pathKeyResolver与配置类中的方法名一致
#            key-resolver: '#{@userKeyResolver}'  #pathKeyResolver与配置类中的方法名一致
#            # 令牌桶每秒填充平均速率
#            redis-rate-limiter.replenishRate: 1
#            # 令牌桶的上限
#            redis-rate-limiter.burstCapacity: 3
        - RewritePath=/product-service/(?<segment>.*), /$\{segment}
      #RequestRateLimiter：使用限流过滤器，是sppringcloud gateway提供的
      #参数 replenishRate:向令牌桶内填充的速率
      #     burstCapacity:令牌桶的容量
#eureka注册中心
eureka:
  client:
    service-url:
      defaultZone: http://localhost:9000/eureka/
  instance:
    prefer-ip-address: true #使用ip地址注册

3.创建配置类GatewayConfiguration

3.1 配置类增加限流规则方法

/**
 * 配置初始化的限流参数
 * 用于指定资源的限流规则
 * 		1.资源名称(yml文件中的路由id)
 * 		2.配置统计时间- setIntervalSec
 * 		3.配置限流阀值
 */
@PostConstruct
public void initGatewayRules() {
	Set<GatewayFlowRule> rules = new HashSet<>();
	rules.add(new GatewayFlowRule("product-service")
			.setCount(1)
			.setIntervalSec(1)
	);

	GatewayRuleManager.loadRules(rules);
}

3.2 修改限流后提示语不明确的配置方法

未增加配置前提示：
	Blocked by Sentinel: ParamFlowException

增加明确方法：

/**
* 自定义一个限流处理器，解决限流后提示信息"Blocked by Sentinel: ParamFlowException"不明确的问题
 */
@PostConstruct
public void initBlockHandlers() {
	BlockRequestHandler blockRequestHandler= new BlockRequestHandler() {
		@Override
		public Mono<ServerResponse> handleRequest(ServerWebExchange serverWebExchange, Throwable throwable) {
			HashMap<Object, Object> map = new HashMap<>();
			map.put("code","200");
			map.put("message","不好意思，限流了");
			return ServerResponse.status(HttpStatus.OK)
					.contentType(MediaType.APPLICATION_JSON_UTF8)
					.body(BodyInserters.fromObject(map));

		}
	};
	GatewayCallbackManager.setBlockHandler(blockRequestHandler);
}

4.自定义分组限流

添加方法：

/**
	 * 自定义API限流分组
	 * 		1.定义分组
	 * 		2.对小组配置限流规则
	 */
	@PostConstruct
	private void initCustomizedApis() {
		Set<ApiDefinition> definitions = new HashSet<>();
		ApiDefinition api1 = new ApiDefinition("product_api")
				.setPredicateItems(new HashSet<ApiPredicateItem>() {{
					add(new ApiPathPredicateItem().setPattern("/product-service/product/**").		//以/product-service/product/开头的所有url
							setMatchStrategy(SentinelGatewayConstants.URL_MATCH_STRATEGY_PREFIX));
				}});
		ApiDefinition api2 = new ApiDefinition("order_api")
				.setPredicateItems(new HashSet<ApiPredicateItem>() {{
					add(new ApiPathPredicateItem().setPattern("/order-service/order"));	//完全匹配/order-service/order的URL
				}});
		definitions.add(api1);
		definitions.add(api2);
		GatewayApiDefinitionManager.loadApiDefinitions(definitions);
	}

修改限流规则方法：

/**
	 * 配置初始化的限流参数
	 * 用于指定资源的限流规则
	 * 		1.资源名称(yml文件中的路由id)
	 * 		2.配置统计时间- setIntervalSec
	 * 		3.配置限流阀值
	 */
	@PostConstruct
	public void initGatewayRules() {
		Set<GatewayFlowRule> rules = new HashSet<>();
		//路由限流
		/*rules.add(new GatewayFlowRule("product-service")
				.setCount(1)
				.setIntervalSec(1)
		);*/

		//自定义API限流分组
		rules.add(new GatewayFlowRule("product_api")
				.setCount(1)
				.setIntervalSec(1));



		GatewayRuleManager.loadRules(rules);
	}

5.限流规则配置类完整代码

package cn.itcast.gateway;

import com.alibaba.csp.sentinel.adapter.gateway.common.SentinelGatewayConstants;
import com.alibaba.csp.sentinel.adapter.gateway.common.api.ApiDefinition;
import com.alibaba.csp.sentinel.adapter.gateway.common.api.ApiPathPredicateItem;
import com.alibaba.csp.sentinel.adapter.gateway.common.api.ApiPredicateItem;
import com.alibaba.csp.sentinel.adapter.gateway.common.api.GatewayApiDefinitionManager;
import com.alibaba.csp.sentinel.adapter.gateway.common.rule.GatewayFlowRule;
import com.alibaba.csp.sentinel.adapter.gateway.common.rule.GatewayParamFlowItem;
import com.alibaba.csp.sentinel.adapter.gateway.common.rule.GatewayRuleManager;
import com.alibaba.csp.sentinel.adapter.gateway.sc.SentinelGatewayFilter;
import com.alibaba.csp.sentinel.adapter.gateway.sc.callback.BlockRequestHandler;
import com.alibaba.csp.sentinel.adapter.gateway.sc.callback.GatewayCallbackManager;
import com.alibaba.csp.sentinel.adapter.gateway.sc.exception.SentinelGatewayBlockExceptionHandler;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.server.ServerResponse;
import org.springframework.web.reactive.result.view.ViewResolver;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.*;

@Configuration
public class GatewayConfiguration {

	private final List<ViewResolver> viewResolvers;

	private final ServerCodecConfigurer serverCodecConfigurer;

	public GatewayConfiguration(ObjectProvider<List<ViewResolver>> viewResolversProvider,
	                            ServerCodecConfigurer serverCodecConfigurer) {
		this.viewResolvers = viewResolversProvider.getIfAvailable(Collections::emptyList);
		this.serverCodecConfigurer = serverCodecConfigurer;
	}

	/**
	 * 配置限流的异常处理器:SentinelGatewayBlockExceptionHandler
	 */
	@Bean
	@Order(Ordered.HIGHEST_PRECEDENCE)
	public SentinelGatewayBlockExceptionHandler sentinelGatewayBlockExceptionHandler() {
		return new SentinelGatewayBlockExceptionHandler(viewResolvers, serverCodecConfigurer);
	}

	/**
	 * 配置限流过滤器
	 */
	@Bean
	@Order(Ordered.HIGHEST_PRECEDENCE)
	public GlobalFilter sentinelGatewayFilter() {
		return new SentinelGatewayFilter();
	}

	/**
	 * 配置初始化的限流参数
	 * 用于指定资源的限流规则
	 * 		1.资源名称(yml文件中的路由id)
	 * 		2.配置统计时间- setIntervalSec
	 * 		3.配置限流阀值
	 */
	@PostConstruct
	public void initGatewayRules() {
		Set<GatewayFlowRule> rules = new HashSet<>();
		//路由限流
		/*rules.add(new GatewayFlowRule("product-service")
				.setCount(1)
				.setIntervalSec(1)
		);*/

		//自定义API限流分组
		rules.add(new GatewayFlowRule("product_api")
				.setCount(1)
				.setIntervalSec(1));



		GatewayRuleManager.loadRules(rules);
	}

	/**
	 * 自定义一个限流处理器，解决限流后提示信息"Blocked by Sentinel: ParamFlowException"不明确的问题
	 */
	@PostConstruct
	public void initBlockHandlers() {
		BlockRequestHandler blockRequestHandler= new BlockRequestHandler() {
			@Override
			public Mono<ServerResponse> handleRequest(ServerWebExchange serverWebExchange, Throwable throwable) {
				HashMap<Object, Object> map = new HashMap<>();
				map.put("code","200");
				map.put("message","不好意思，限流了");
				return ServerResponse.status(HttpStatus.OK)
						.contentType(MediaType.APPLICATION_JSON_UTF8)
						.body(BodyInserters.fromObject(map));

			}
		};
		GatewayCallbackManager.setBlockHandler(blockRequestHandler);
	}

	/**
	 * 自定义API限流分组
	 * 		1.定义分组
	 * 		2.对小组配置限流规则
	 */
	@PostConstruct
	private void initCustomizedApis() {
		Set<ApiDefinition> definitions = new HashSet<>();
		ApiDefinition api1 = new ApiDefinition("product_api")
				.setPredicateItems(new HashSet<ApiPredicateItem>() {{
					add(new ApiPathPredicateItem().setPattern("/product-service/product/**").		//以/product-service/product/开头的所有url
							setMatchStrategy(SentinelGatewayConstants.URL_MATCH_STRATEGY_PREFIX));
				}});
		ApiDefinition api2 = new ApiDefinition("order_api")
				.setPredicateItems(new HashSet<ApiPredicateItem>() {{
					add(new ApiPathPredicateItem().setPattern("/order-service/order"));	//完全匹配/order-service/order的URL
				}});
		definitions.add(api1);
		definitions.add(api2);
		GatewayApiDefinitionManager.loadApiDefinitions(definitions);
	}
}

三.详情可下载学习文档：

学习文档下载