分布式系列之限流组件

2024-06-20 16:20
文章标签 组件 系列 分布式 限流

本文主要是介绍分布式系列之限流组件，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

概述

在高并发场景下，请求量瞬间到达，后端服务器即使有缓存、集群主备、分库分表、容错降级等措施，也有可能扛不住这请求量，因此可考虑引入限流组件。限流的目的：防止恶意请求流量或流量超出系统承载。

应用场景：

  • 网关层校验流量，拦截非法请求，或直接抛弃部分流量（后来的流量，如秒杀系统）
  • 实时场景下的数据迁移或复制（如Kafka分区重分配）

本文从限流算法、限流环节、限流框架几个层次加以讲述。

限流算法

主要有两大类：窗口算法（也有叫做计数器算法的，思想其实很类似，包括固定窗口算法和滑动窗口算法）、桶算法（包括漏桶算法、令牌桶算法）。

计数器算法

该算法会维护一个counter，规定在单位时间内counter的大小不能超过最大值，每隔固定时间就将counter的值置零。如果这个counter大于设定的阈值，那么系统就开始拒绝请求以保护系统的负载。

固定窗口

固定窗口就是指定的单位时间，比如一分钟，限制次数为Max，则最大值为Max，超过Max的请求被抛弃。存在临界问题：单位时间的左区间可能有大量请求，直接超过Max。

/*** 指定过期时间自增计数器，默认每次+1，非滑动窗口** @param key 计数器自增key* @param expireTime 过期时间* @param unit  时间单位*/
public long incrCount(String key, int expireTime, TimeUnit unit);/*** 指定过期时间自增计数器,单位时间内超过最大值rateThreshold返回true，否则返回false** @param key 限流key* @param rateThreshold 限流阈值* @param expireTime 固定窗口时间* @param unit 时间单位*/
public boolean rateLimit(final String key, final int rateThreshold, int expireTime, TimeUnit unit);

滑动窗口

固定窗口可看成是滑动窗口的特例。滑动窗口将固定窗口再等分为多个小的窗口，每一次对一个小的窗口进行流量控制。可解决固定窗口的临界问题。

漏桶算法

Leaky Bucket，水（请求）先进入到漏桶（预先维护的容量固定）里，漏桶以一定的速度出水（接口有响应速率），当水流入速度过大会直接溢出（访问频率超过接口响应速率），然后就拒绝请求，漏桶算法能强行限制数据的传输速率。
在这里插入图片描述
核心：请求来以后，直接进桶，然后桶根据自己的漏洞大小慢慢往外面漏。使用一个FIFO的队列，一端负责不断的放入请求，另外一端负责吐出请求。

好处：可将系统的处理能力维持在一个比较平稳的水平
缺点：不能应对实际场景，比如突然暴增的流量。

令牌桶算法

Token Bucket Algorithm，用于限制速率。

令牌桶会以一个恒定的速率（预定义）向固定容量（要限制流量的大小）大小桶中放入令牌，当有流量来时则取走一个或多个令牌。当桶中没有令牌则将当前请求丢弃或阻塞等待，直到有足够的令牌为止。可实现平滑突发限流，允许一定的突发请求，但总体速率会保持在设定的限制内。
在这里插入图片描述
参考实现：Guava RateLimiter

漏桶算法和令牌桶算法对比

令牌桶算法和漏桶算法的不同之处在于处理瞬间到达的大流量的不同：

  • 令牌桶算法由于在令牌桶里攒很多令牌，因此在大流量到达一瞬间可以一次性将队列中所有的请求都处理完，然后按照恒定的速度处理请求；
  • 漏桶算法则一直有一个恒等的阈值，在大流量到达时，也会将多余的请求拒绝。

限流环节

限流从环节（也有叫粒度）来分析，有：网关限流，服务限流，接口限流。

网关限流

网关，即接入层，是请求流量的入口，一般可考虑使用Nginx限流。

Nginx自带两个模块：

  • 连接数限流模块：ngx_http_limit_conn_module（简称limit_conn，如果读者们在某些书籍或Blog里看到这个简称，就表示这个模块）
  • 请求限流模块：基于漏桶算法实现，ngx_http_limit_req_module（简称limit_req，同上）。

此外，可以使用OpenResty提供的lua限流模块lua-resty-limit-traffic，以及tengine增强版httplimitreqcn。

limit_conn

limit_conn模块用于限制连接数量，特别是来自单个IP地址的连接数量。并非所有的连接都被计数。只有当服务器处理完请求且已经读取整个请求头时，连接才被计数。

官网的示例配置：

http {limit_conn_zone $binary_remote_addr zone=addr:10m;server {location /download/ {limit_conn addr 1;}	}
}

配置解读：
limit_conn_zone $binary_remote_addr zone=addr:10m;：为共享内存区域设置参数，该区域将保留各种Key键的状态。状态包含当前的连接数。Key可以包含文本，变量，他们的组合。只能用于http语法块。

$binary_remote_addr：对于IPv4地址，变量的大小始终为4个字节，对于IPv6地址则为16个字节。存储状态在32位平台上始终占用32或64个字节，在64位平台上占用64个字节。一个兆字节的区域可保存大约32000个32字节的状态或大约16000个64字节的状态。如果区域存储耗尽，服务器会将错误返回给所有其他请求。10M可存储160000个状态。

limit_conn addr 1：设置给定键值的共享内存区域和最大允许连接数。超过此限制时，服务器将返回503错误以回复请求。用于http，server，location这些语法块内。

limit_req

用于限制每一个请求的处理速率，即每个IP地址的请求的处理速率。

官网的示例配置：

http {limit_req_zone $binary_remote_addr zone=one:10m rate=1r/s;server {location /search/ {limit_req zone=one burst=5;}
}

配置解读：
limit_req_zone $binary_remote_addr zone=one:10m rate=1r/s;：区域名称为one，大小为10m，平均处理的请求频率不能超过每秒一次。键值是客户端IP，$binary_remote_addr变量，可以将每条状态记录的大小减少到64个字节，这样1M的内存可以保存大约1万6千个64字节的记录。如果限制域的存储空间耗尽，对于后续所有请求，服务器都会返回503错误。

limit_req zone=one burst=5;：平均每秒不超过1个请求，并且突发不超过5个请求。超过处理能力范围的，直接drop，表现为对收到的请求无延时。

服务限流

应用级别，对应用里涉及到的所有的接口类和方法都增加限流机制。

接口限流

接口层次的限流，粒度最小，用于某个Controller类或方法，

限流框架

Guava RateLimiter

Guava提供的抽象类RateLimiter源码很长，略。用于限制方法的调用频率，它基于令牌桶算法

其实现类有两个，对应两种限流模式：

  1. SmoothBursty：稳定模式，令牌生成速度恒定，平滑突发限流；
  2. SmoothWarmingUp：渐进模式，令牌生成速度缓慢提升直到维持在一个稳定值，平滑预热限流

两种模式实现思路类似，主要区别在等待时间的计算上。

局限性：

  1. 仅适用于单体应用
  2. 不保证公平性访问

Spring Cloud GateWay

需要引入依赖：

<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-gateway</artifactId>
</dependency>
<!--基于 reactive stream 的redis -->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>

引入spring-cloud-starter-gateway后，则引入子依赖spring-cloud-gateway-server，后者源码里带有request_rate_limiter.lua脚本：
在这里插入图片描述
配置：

spring:cloud:gateway:routes:- id: requestratelimiter_routeuri: lb://gateway-demoorder: 10000predicates:- Path = /admin/**filters:- name: RequestRateLimiterargs:redis-rate-limiter.replenishRate: 1 # 流速每秒1个redis-rate-limiter.burstCapacity: 3 # 令牌桶容积key - resolver: "#{@remoteAddrKeyResolver}" # SPEL表达式去的对应的bean- StripPrefix = 1

配置Bean方法，多维度限流量的入口：

@Bean
KeyResolver remoteAddrKeyResolver() {return exchange -> Mono.just(exchange.getRequest().getRemoteAddress().getHostName());
}

GateWay提供RateLimiter接口，AbstractRateLimiter抽象类，以及实现类RedisRateLimiter，接口主要有一个方法Mono<Response> isAllowed(String routeId, String id);，RedisRateLimter核心源码：

public Mono<Response> isAllowed(String routeId, String id) {Config routeConfig = getConfig().getOrDefault(routeId, defaultConfig);int replenishRate = routeConfig.getReplenishRate();int burstCapacity = routeConfig.getBurstCapacity();try {List<String> keys = getKeys(id);/* returns unixtime in seconds.*/List<String> scriptArgs = Arrays.asList(replenishRate + "", burstCapacity + "", Instant.now().getEpochSecond() + "", "1");// 核心,执行redis LUA脚本Flux<List<Long>> flux = this.redisTemplate.execute(this.script, keys, scriptArgs);return flux.onErrorResume(throwable -> Flux.just(Arrays.asList(1L, -1L))).reduce(new ArrayList<Long>(), (longs, l) -> {longs.addAll(l);return longs;}).map(results -> {boolean allowed = results.get(0) == 1L;Long tokensLeft = results.get(1);Response response = new Response(allowed, getHeaders(routeConfig, tokensLeft));if (log.isDebugEnabled()) {log.debug("response: " + response);}return response;});} catch (Exception e) {log.error("Error determining if user allowed from redis", e);}return Mono.just(new Response(true, getHeaders(routeConfig, -1L)));
}

Sentinel

另起一篇，参考服务容错框架Sentinel入门。

Bucket4j

Bucket4j是一个基于令牌桶算法实现的强大的限流库，支持单机限流，还支持通过诸如Hazelcast、Ignite、Coherence、Infinispan或其他兼容 JCache API(JSR 107)规范的分布式缓存实现分布式限流。

3个核心概念：Bucket、Bandwidth、Refill。

Bucket：接口，代表令牌桶的具体实现，操作的入口。提供诸如tryConsume和tryConsumeAndReturnRemaining方法用于消费令牌。可通过下面的构造方法来创建Bucket：

Bucket bucket = Bucket4j.builder().addLimit(limit).build();
if (bucket.tryConsume(1)) {log.info("ok");
} else {log.info("error");
}

Bandwidth：带宽，可理解为限流规则。Bucket4j提供两种方法来创建Bandwidth：

  • simple：桶大小和填充速度是一样的，表示桶大小为10，填充速度为每分钟10个令牌：
Bandwidth limit = Bandwidth.simple(10, Duration.ofMinutes(1));
  • classic：更灵活，可自定义填充速度：
// 贪婪策略：桶大小为10，填充速度为每分钟5个令牌;
Refill filler = Refill.greedy(5, Duration.ofMinutes(1));
Bandwidth limit = Bandwidth.classic(10, filler);

Refill：用于填充令牌桶，可以通过它定义填充速度，Bucket4j有两种填充令牌的策略：

  • intervally：间隔策略。间隔策略创建Refill：Refill filler = Refill.intervally(5, Duration.ofMinutes(1));// 每隔一分钟，填充 5 个令牌
    所谓间隔策略指的是每隔一段时间，一次性的填充所有令牌。
  • greedy：贪婪策略，参考上面的简单例子

Resilience4j

Resilience4j是一款轻量级、易使用的高可用框架，设计灵感就来自于Netflix Hystrix。自从Hystrix停止维护之后，官方也推荐使用Resilience4j来代替Hystrix。

Resilience4j的底层采用Vavr（一个非常轻量级的Java函数式库），以装饰器模式提供对函数式接口或lambda表达式的封装，提供高可用机制：重试（Retry）、熔断（Circuit Breaker）、限流（RateLimiter）、限时（Timer Limiter）、隔离（Bulkhead）、缓存（Cache）和降级（Fallback）。

其中，RateLimiter是请求频率限流，Bulkhead是并发量限流。

Resilience4j提供两种限流的实现：SemaphoreBasedRateLimiter和AtomicRateLimiter。SemaphoreBasedRateLimiter基于信号量实现，用户的每次请求都会申请一个信号量，并记录申请的时间，申请通过则允许请求，申请失败则限流，另外有一个内部线程会定期扫描过期的信号量并释放，很显然这是令牌桶的算法。AtomicRateLimiter和上面的经典实现类似，不需要额外的线程，在处理每次请求时，根据距离上次请求的时间和生成令牌的速度自动填充。

Resilience4j也提供两种隔离的实现：SemaphoreBulkhead和ThreadPoolBulkhead，通过信号量或线程池控制请求的并发数。

同时使用限流和隔离的例子：

// 创建一个Bulkhead,最大并发量为150
BulkheadConfig bulkheadConfig = BulkheadConfig.custom().maxConcurrentCalls(150).maxWaitTime(100).build();
Bulkhead bulkhead = Bulkhead.of("backendName", bulkheadConfig);// 创建RateLimiter，每秒允许一次请求
RateLimiterConfig rateLimiterConfig = RateLimiterConfig.custom().timeoutDuration(Duration.ofMillis(100)).limitRefreshPeriod(Duration.ofSeconds(1)).limitForPeriod(1).build();
RateLimiter rateLimiter = RateLimiter.of("backendName", rateLimiterConfig);// 使用Bulkhead和RateLimiter装饰业务逻辑
Supplier<String> supplier = () -> backendService.doSomething();
Supplier<String> decoratedSupplier = Decorators.ofSupplier(supplier).withBulkhead(bulkhead).withRateLimiter(rateLimiter).decorate();// 调用业务逻辑
Try<String> try = Try.ofSupplier(decoratedSupplier);
assertThat(try.isSuccess()).isTrue();

Resilience4j在功能特性上比Bucket4j强大不少，而且还支持并发量限流，但不支持分布式限流。

拓展

TimeLimiter和RateLimiter

除了RateLimiter，在Guava包里还有个TimeLimiter接口：

@DoNotMock("Use FakeTimeLimiter")
@ElementTypesAreNonnullByDefault
@J2ktIncompatible
@GwtIncompatible
public interface TimeLimiter {<T> T newProxy(T target, Class<T> interfaceType, long timeoutDuration, TimeUnit timeoutUnit);default <T> T newProxy(T target, Class<T> interfaceType, Duration timeout) {return this.newProxy(target, interfaceType, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);}@ParametricNullness@CanIgnoreReturnValue<T> T callWithTimeout(Callable<T> callable, long timeoutDuration, TimeUnit timeoutUnit) throws TimeoutException, InterruptedException, ExecutionException;@ParametricNullness@CanIgnoreReturnValuedefault <T> T callWithTimeout(Callable<T> callable, Duration timeout) throws TimeoutException, InterruptedException, ExecutionException {return this.callWithTimeout(callable, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);}@ParametricNullness@CanIgnoreReturnValue<T> T callUninterruptiblyWithTimeout(Callable<T> callable, long timeoutDuration, TimeUnit timeoutUnit) throws TimeoutException, ExecutionException;@ParametricNullness@CanIgnoreReturnValuedefault <T> T callUninterruptiblyWithTimeout(Callable<T> callable, Duration timeout) throws TimeoutException, ExecutionException {return this.callUninterruptiblyWithTimeout(callable, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);}void runWithTimeout(Runnable runnable, long timeoutDuration, TimeUnit timeoutUnit) throws TimeoutException, InterruptedException;default void runWithTimeout(Runnable runnable, Duration timeout) throws TimeoutException, InterruptedException {this.runWithTimeout(runnable, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);}void runUninterruptiblyWithTimeout(Runnable runnable, long timeoutDuration, TimeUnit timeoutUnit) throws TimeoutException;default void runUninterruptiblyWithTimeout(Runnable runnable, Duration timeout) throws TimeoutException {this.runUninterruptiblyWithTimeout(runnable, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);}
}

其实现类有：

  • SimpleTimeLimiter：
  • FakeTimeLimiter：

TimeLimiter用于限制方法的执行时间，原理：

  • 创建代理对象：为指定的对象创建一个代理对象
  • 方法拦截：当调用代理对象的方法时，TimeLimiter会拦截调用，并在一个单独的线程中执行实际方法
  • 超时监控：使用定时器来监控方法的执行时间。如果方法执行超过指定的时间限制，TimeLimiter会中断该线程，并抛出一个TimeoutException

TimeLimiter可确保方法不会因为执行时间过长而阻塞其他操作。有如下使用场景：

  • 网络调用：限制网络请求的最大时间，防止由于网络延迟导致的长时间阻塞
  • 复杂计算：在后台执行复杂计算任务时，设置一个时间限制以确保任务不会无限期地运行
  • 资源获取：限制资源获取操作的时间，如数据库连接、文件读写等。

TimeLimiter的一个简单实例：

public void test() {TimeLimiter timeLimiter = SimpleTimeLimiter.create();Callable<String> callable = () -> {// Simulate long running taskThread.sleep(2000);return "Task Completed";};try {String result = timeLimiter.callWithTimeout(callable, 1, TimeUnit.SECONDS);System.out.println(result);} catch (UncheckedTimeoutException e) {// just logging or handle exception} catch (Exception e) {// just logging or handle exception}
}

TimeLimiter和RateLimiter的区别

  • 用途：TimeLimiter用于限制方法的执行时间，防止长时间运行的操作阻塞程序。RateLimiter用于限制方法的调用频率，控制流量，防止系统过载
  • 实现机制：TimeLimiter基于创建代理对象和超时监控，通过定时器中断执行时间过长的任务。RateLimiter基于令牌桶算法，通过生成和消耗令牌来控制请求的执行频率。
  • 使用场景：TimeLimiter适用于需要对单个任务的执行时间进行严格控制的场景。RateLimiter适用于需要对系统整体调用频率进行控制的场景。

Kafka

Kafka在时使用到复制限流技术，防止集群中某个主题或某个分区的流量在某段时间内特别大，分区数据复制造成数据丢失或分区节点异常。

ZooKeeper

ZK也提供一个RateLimiter的简易实现版本。

分布式限流

参考

  • 亿级流量网站架构核心技术
  • 实战Spring Cloud Gateway之限流篇

这篇关于分布式系列之限流组件的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1078634。

相关文章

vue解决子组件样式覆盖问题scoped deep

vue解决子组件样式覆盖问题scoped deep

《vue解决子组件样式覆盖问题scopeddeep》文章主要介绍了在Vue项目中处理全局样式和局部样式的方法,包括使用scoped属性和深度选择器（/deep/）来覆盖子组件的样式,作者建议所有组件... 目录前言scoped分析deep分析使用总结所有组件必须加scoped父组件覆盖子组件使用deep前言
阅读更多...
基于Qt Qml实现时间轴组件

基于Qt Qml实现时间轴组件

《基于QtQml实现时间轴组件》时间轴组件是现代用户界面中常见的元素,用于按时间顺序展示事件,本文主要为大家详细介绍了如何使用Qml实现一个简单的时间轴组件,需要的可以参考下... 目录写在前面效果图组件概述实现细节1. 组件结构2. 属性定义3. 数据模型4. 事件项的添加和排序5. 事件项的渲染如何使用
阅读更多...
java如何分布式锁实现和选型

java如何分布式锁实现和选型

《java如何分布式锁实现和选型》文章介绍了分布式锁的重要性以及在分布式系统中常见的问题和需求,它详细阐述了如何使用分布式锁来确保数据的一致性和系统的高可用性,文章还提供了基于数据库、Redis和Zo... 目录引言：分布式锁的重要性与分布式系统中的常见问题和需求分布式锁的重要性分布式系统中常见的问题和需求
阅读更多...
Golang使用etcd构建分布式锁的示例分享

Golang使用etcd构建分布式锁的示例分享

《Golang使用etcd构建分布式锁的示例分享》在本教程中,我们将学习如何使用Go和etcd构建分布式锁系统,分布式锁系统对于管理对分布式系统中共享资源的并发访问至关重要,它有助于维护一致性,防止竞... 目录引言环境准备新建Go项目实现加锁和解锁功能测试分布式锁重构实现失败重试总结引言我们将使用Go作
阅读更多...
Redis分布式锁使用及说明

Redis分布式锁使用及说明

《Redis分布式锁使用及说明》本文总结了Redis和Zookeeper在高可用性和高一致性场景下的应用,并详细介绍了Redis的分布式锁实现方式,包括使用Lua脚本和续期机制,最后,提到了RedLo... 目录Redis分布式锁加锁方式怎么会解错锁？举个小案例吧解锁方式续期总结Redis分布式锁如果追求
阅读更多...
基于Redis有序集合实现滑动窗口限流的步骤

基于Redis有序集合实现滑动窗口限流的步骤

《基于Redis有序集合实现滑动窗口限流的步骤》滑动窗口算法是一种基于时间窗口的限流算法,通过动态地滑动窗口,可以动态调整限流的速率,Redis有序集合可以用来实现滑动窗口限流,本文介绍基于Redis... 滑动窗口算法是一种基于时间窗口的限流算法，它将时间划分为若干个固定大小的窗口，每个窗口内记录了该时间
阅读更多...
Spring Security 从入门到进阶系列教程

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门（一）：登录与退出》 《Spring-Security-入门（二）：基于数据库验证》 《Spring-Security-入门（三）：密码加密》 《Spring-Security-入门（四）：自定义-Filter》 《Spring-Security-入门（五）：在 Sprin
阅读更多...
JS常用组件收集

JS常用组件收集

收集了一些平时遇到的前端比较优秀的组件，方便以后开发的时候查找！！！ 函数工具： Lodash 页面固定： stickUp、jQuery.Pin 轮播： unslider、swiper 开关: switch 复选框： icheck 气泡： grumble 隐藏元素： Headroom
阅读更多...
如何在页面调用utility bar并传递参数至lwc组件

如何在页面调用utility bar并传递参数至lwc组件

1.在app的utility item中添加lwc组件： 2.调用utility bar api的方式有两种： 方法一，通过lwc调用： import {LightningElement,api ,wire } from 'lwc';import { publish, MessageContext } from 'lightning/messageService';import Ca
阅读更多...
科研绘图系列：R语言扩展物种堆积图（Extended Stacked Barplot）

科研绘图系列：R语言扩展物种堆积图（Extended Stacked Barplot）

介绍 R语言的扩展物种堆积图是一种数据可视化工具，它不仅展示了物种的堆积结果，还整合了不同样本分组之间的差异性分析结果。这种图形表示方法能够直观地比较不同物种在各个分组中的显著性差异，为研究者提供了一种有效的数据解读方式。 加载R包 knitr::opts_chunk$set(warning = F, message = F)library(tidyverse)library(phyl
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2