限流

一 背景在零碎高可用设计中，接口限流是一个十分重要环节，一方面是出于对本身服务器资源的爱护，另一方面也是对依赖资源的一种保护措施。比方对于 Web 利用，我限度单机只能解决每秒 1000 次的申请，超过的局部间接返回谬误给客户端。尽管这种做法侵害了用户的应用体验，然而它是在极其并发下的无奈之举，是短暂的行为，因而是能够承受的。 二 设计思路常见的限流有2种思路 第一种是限度总量,也就是限度某个指标的累积下限，常见的是限度以后零碎服务的用户总量，例如：某个抢购流动商品数量只有 100 个，限度参加抢购的用户下限为 1 万个，1 万当前的用户间接回绝。第二种是限度工夫量，也就是限度一段时间内某个指标的下限，例如 1 分钟内只容许 10000 个用户拜访；每秒申请峰值最高为 10 万。三 限流算法目前实现限流算法次要分为3类，这里不具体开展介绍： 1）工夫窗口 固定工夫窗口算法是最简略的限流算法，它的实现原理就是管制单位工夫内申请的数量，然而这个算法有个毛病就是临界值问题。 为了解决临界值的问题，又推出滑动工夫窗口算法，其实现原理大抵上是将工夫分为一个一个小格子，在统计申请数量的时候，是通过统计滑动工夫周期内的申请数量。 2）漏斗算法 漏斗算法的外围是管制总量，申请流入的速率不确定，超过流量局部益出，该算法比拟实用于针对突发流量，想要尽可能的接管全副申请的场景。其毛病也比拟显著，这个总量怎么评估，大小怎么配置，而且一旦初始化也没法动静调整。 3）令牌桶算法 令牌桶算法的外围是管制速率，令牌产生的速度是要害，一直的申请获取令牌，获取不到就抛弃。该算法比拟实用于针对突发流量，以爱护本身服务资源以及依赖资源为主，反对动静调整速率。毛病的话实现比较复杂，而且会抛弃很多申请。 四 实现步骤咱们自定义的这套限流组件有是基于guava RateLimiter封装的，采纳令牌桶算法以管制速率为主，反对DUCC动静配置，同时反对限流后的降级措施。接下来看一下整体实现计划 1、自定义RateLimiter Annotation标签这里次要对限流相干属性的一个定义，包含每秒产生的令牌数、获取令牌超时工夫、降级逻辑实现以及限流开关等内容 @Documented@Target({ElementType.METHOD, ElementType.TYPE})@Retention(RetentionPolicy.RUNTIME)public @interface SysRateLimit { /** * 每秒产生的令牌数 默认500 * * @return */ double permitsPerSecond() default 500D; /** * 获取令牌超时工夫 默认100 * * @return */ long timeout() default 100; /** * 获取令牌超时工夫单位 默认毫秒 * * @return */ TimeUnit timeUnit() default TimeUnit.MILLISECONDS; /** * 服务降级办法名称 Spring bean id * * @return */ String fallbackBeanId() default ""; /** * 限流key 惟一 * * @return */ String limitKey() default "";}2、基于Spring Aspect 结构切面首先就是咱们须要结构一个Aspect切面用于扫描咱们自定义的SysRateLimit标签 ...

作者：京东科技 康志兴利用场景古代互联网很多业务场景，比方秒杀、下单、查问商品详情，最大特点就是高并发，而往往咱们的零碎不能接受这么大的流量，继而产生了很多的应答措施：CDN、音讯队列、多级缓存、异地多活。 然而无论如何优化，究竟由硬件的物理个性决定了咱们零碎性能的下限，如果强行接管所有申请，往往造成雪崩。 这时候限流熔断就发挥作用了，限度申请数，疾速失败，保证系统满负载又不超限。 极致的优化，就是将硬件使用率进步到100%，但永远不会超过100%罕用限流算法1. 计数器间接计数，简略暴力，举个例子： 比方限流设定为1小时内10次，那么每次收到申请就计数加一，并判断这一小时内计数是否大于下限10，没超过下限就返回胜利，否则返回失败。 这个算法的毛病就是在工夫临界点会有较大霎时流量。 持续下面的例子，现实状态下，申请匀速进入，零碎匀速解决申请： 但理论状况中，申请往往不是匀速进入，假如第n小时59分59秒的时候忽然进入10个申请，全副申请胜利，达到下一个工夫区间时刷新计数。那么第n+1小时刚开始又打进10个申请，等于霎时进入20个申请，必定不合乎“1小时10次”的规定，这种景象叫做“突刺景象”。 为解决这个问题，计数器算法通过优化后，产生了滑动窗口算法： 咱们将工夫距离平均分隔，比方将一分钟分为6个10秒，每一个10秒内独自计数，总的数量限度为这6个10秒的总和，咱们把这6个10秒成为“窗口”。 那么每过10秒，窗口往前滑动一步，数量限度变为新的6个10秒的总和，如图所示： 那么如果在临界时，收到10个申请（图中灰色格子），在下一个时间段来长期，橙色局部又进入10个申请，但窗口内蕴含灰色局部，所以曾经达到申请上线，不再接管新的申请。 这就是滑动窗口算法。 然而滑动窗口依然有缺点，为了保障匀速，咱们要划分尽可能多的格子，而格子越多，每一个格子可能接管的申请数就越少，这样就限度了零碎霎时解决能力。 2. 漏桶 漏桶算法其实也很简略，假如咱们有一个固定容量的桶，流速（零碎解决能力）固定，如果一段时间水龙头水流太大，水就溢出了（申请被抛弃了）。 用编程的语言来说，每次申请进来都放入一个先进先出的队列中，队列满了，则间接返回失败。另外有一个线程池固定距离一直地从这个队列中拉取申请。 音讯队列、jdk的线程池，都有相似的设计。 3. 令牌桶令牌桶算法比漏桶算法稍显简单。 首先，咱们有一个固定容量的桶，桶里寄存着令牌（token）。桶一开始是空的，token以一个固定的速率往桶里填充，直到达到桶的容量，多余的令牌将会被抛弃。每当一个申请过去时，就会尝试从桶里移除一个令牌，如果没有令牌的话，申请无奈通过。 漏桶和令牌桶算法的区别：漏桶的特点是生产能力固定，当申请量超出生产能力时，提供肯定的冗余能力，把申请缓存下来匀速生产。长处是对上游爱护更好。 令牌桶遇到激增流量会更从容，只有存在令牌，则能够一并生产掉。适宜有突发特色的流量，如秒杀场景。 限流计划一、容器限流1. Tomcattomcat可能配置连接器的最大线程数属性，该属性maxThreads是Tomcat的最大线程数，当申请的并发大于maxThreads时，申请就会排队执行(排队数设置：accept-count)，这样就实现了限流的目标。 <Connector port="8080" protocol="HTTP/1.1" connectionTimeout="20000" maxThreads="150" redirectPort="8443" />2. NginxNginx 提供了两种限流伎俩：一是管制速率，二是管制并发连接数。 管制速率 咱们须要应用limit_req_zone配置来限度单位工夫内的申请数，即速率限度，示例配置如下： limit_req_zone $binary_remote_addr zone=mylimit:10m rate=2r/s;第一个参数：$binary\_remote\_addr 示意通过remote\_addr这个标识来做限度，“binary\_”的目标是缩写内存占用量，是限度同一客户端ip地址。 第二个参数：zone=mylimit:10m示意生成一个大小为10M，名字为one的内存区域，用来存储拜访的频次信息。 第三个参数：rate=2r/s示意容许雷同标识的客户端的拜访频次，这里限度的是每秒2次，还能够有比方30r/m的。 并发连接数 利用limit_conn_zone和limit_conn两个指令即可管制并发数，示例配置如下 limit_conn_zone $binary_remote_addr zone=perip:10m;limit_conn_zone $server_name zone=perserver:10m;server { ... limit_conn perip 10; # 限度同一个客户端ip limit_conn perserver 100;}只有当 request header 被后端解决后，这个连贯才进行计数 二、服务端限流1. SemaphoreJUC包中提供的信号量工具，它的外部保护了一个同步队列，咱们能够在每个申请进来的时候，尝试获取信号量，获取不到能够阻塞或者疾速失败 简略样例： Semaphore sp = new Semaphore(3);sp.require(); // 阻塞获取System.out.println("执行业务逻辑");sp.release();2. RateLimiterGuava中基于令牌桶实现的一个限流工具，应用非常简单，通过办法create()创立一个桶，而后通过acquire()或者tryAcquire()获取令牌： ...

1. 限流的意义明天谈谈限流。很早之前，在接触像 hystrix、resilience4j、sentinel 这类的熔断器组件时，就理解过其对于限流的性能。在理论开发利用中，超时、谬误熔断用的挺多，但限流熔断用的到不多。 究其原因，在公司外部微服务调用时，就算服务调用的上下游服务，不是同一个我的项目团队的服务，但至多是同一个公司的研发团队。当避免上游方被频繁调用，齐全能够和上游方约定好协同计划，而不是通过限流的策略给上游方抛错。 但如果上下游方比拟独立，则有必要通过限流来进行束缚，和自我爱护。 例如：咱们产品对外提供的服务端凋谢API，如果不在文档中约定好调用频率限度，并做好自我爱护，很容易就被人歹意攻打。 再例如：咱们在对接钉钉、微信等生态服务时，也须要调用它们在开放平台的API，同样也无限流要求。可参考 钉钉服务端API限流文档。因为一旦调用钉钉API频率超限，会触发至多5分钟的限流熔断，这5分钟内任何API调用都会报错。所以咱们作为服务调用方，更要将调用服务的申请限流。 2. 限流算法在网上看了些限流算法，次要有4种。 2.1. 固定窗口算法首先保护一个计数器，将单位时间段当做一个窗口，计数器记录这个窗口接管申请的次数。 当次数少于限流阀值，就容许拜访，并且计数器+1当次数大于限流阀值，就回绝拜访。以后的工夫窗口过来之后，计数器清零。假如单位工夫是1秒，限流阀值为3。在单位工夫1秒内，每来一个申请,计数器就加1，如果计数器累加的次数超过限流阀值3，后续的申请全副回绝。等到1s完结后，计数器清0，从新开始计数。 问题 1：窗口临界值，导致双倍阈值假如限流阀值为5个申请，单位工夫窗口是1s,如果咱们在单位工夫内的前0.8-1s和1-1.2s，别离并发5个申请。尽管都没有超过阀值，然而如果算0.8-1.2s,则并发数高达10，曾经超过单位工夫1s不超过5阀值的定义啦，通过的申请达到了阈值的两倍。 为了解决问题2中窗口临界值的问题，引入了滑动窗口限流。滑动窗口限流解决固定窗口临界值的问题，能够保障在任意工夫窗口内都不会超过阈值。 问题 2（两面性）：集中流量，打满阈值，后续服务不可用比方窗口大小为1s，限流大小为100，而后恰好在某个窗口的第1ms来了100个申请，而后第2ms-999ms的申请就都会被回绝，这段时间用户会感觉零碎服务不可用。 但这是两面性的问题，有毛病，也有长处，前面会说。 2.2. 滑动窗口算法绝对于固定窗口，滑动窗口除了须要引入计数器之外，还须要记录时间窗口内每个申请达到的工夫点，因而对内存的占用会比拟多。 规定如下，假如工夫窗口为 1 秒： 记录每次申请的工夫。统计每次申请的工夫 至 往前推1秒这个工夫窗口内申请数，并且 1 秒前的数据能够删除。统计的申请数小于阈值就记录这个申请的工夫，并容许通过，反之回绝。滑动窗口算法就是固定窗口的升级版。将计时窗口划分成一个小窗口，滑动窗口算法就进化成了固定窗口算法。而滑动窗口算法其实就是对申请数进行了更细粒度的限流，窗口划分的越多，则限流越精准。 然而滑动窗口和固定窗口都无奈解决短时间之内集中流量的突击，就和后面介绍的一样。 接下来再说说漏桶，它能够解决工夫窗口类的痛点，使得流量更加的平滑。 2.3. 漏桶算法漏桶算法面对限流，就更加的柔性，不存在间接的粗犷回绝。 它的原理很简略，能够认为就是注水漏水的过程。往漏桶中以任意速率流入水，以固定的速率流出水。当水超过桶的容量时，会被溢出，也就是被抛弃。因为桶容量是不变的，保障了整体的速率。 流入的水滴，能够看作是拜访零碎的申请，这个流入速率是不确定的。桶的容量个别示意零碎所能解决的申请数。如果桶的容量满了，就达到限流的阀值，就会抛弃水滴（拒绝请求）流出的水滴，是恒定过滤的，对应服务依照固定的速率解决申请。看到这想到啥，是不是 和音讯队列思维有点像，削峰填谷。通过破绽这么一过滤，申请就能平滑的流出，看起来很像很挺完满的？实际上它的长处也即毛病。 问题 3: 无奈应答流量突发面对突发申请，服务的处理速度和平时是一样的，这其实不是咱们想要的，在面对突发流量咱们心愿在零碎安稳的同时，晋升用户体验即能更快的解决申请，而不是和失常流量一样，安分守己的解决（看看，之前滑动窗口说流量不够平滑，当初太平滑了又不行，难搞啊）。 而接下来咱们要谈的令牌桶算法可能在肯定水平上解决流量突发的问题。 2.4. 令牌桶算法令牌桶算法是对漏斗算法的一种改良，除了可能起到限流的作用外，还容许肯定水平的流量突发。 令牌桶算法原理： 有一个令牌管理员，依据限流大小，定速往令牌桶里放令牌。如果令牌数量满了，超过令牌桶容量的限度，那就抛弃。零碎在承受到一个用户申请时，都会先去令牌桶要一个令牌。如果拿到令牌，那么就解决这个申请的业务逻辑；如果拿不到令牌，就间接回绝这个申请。能够看出令牌桶在应答突发流量的时候，桶内如果有 100 个令牌，那么这 100 个令牌能够马上被取走，而不像漏桶那样匀速的生产。所以在应答突发流量的时候令牌桶体现的更佳。 2.5. 集体了解依照我的了解，对这4种算法做上面的分类比拟。 2.5.1. 滑动窗口算法 > 固定窗口算法固定窗口算法实现简略，性能高。然而会有临界突发流量问题，刹时流量最大能够达到阈值的2倍。 为了解决临界突发流量，能够将窗口划分为多个更细粒度的单元，每次窗口向右挪动一个单元，于是便有了滑动窗口算法。 从算法成果上来讲滑动窗口算法要优于固定窗口算法，毕竟能防止窗口临界值问题。 从施行性能上来讲固定窗口算法实现起来要更简略，对性能资源要求更低。滑动窗口只须要引入计数器，但滑动窗口还须要记录时间窗口内每个申请达到的工夫点，因而对内存的占用会比拟多。 总结：滑动窗口算法优先不过限流算法就是为了爱护线上服务器资源，防止被流量击溃。与这代价相比，滑动窗口算法的那些性能资源耗费算得了什么。所以目前市场上，简直看不到以固定窗口算法实现的限流组件。 2.5.2. 漏桶算法（MQ音讯队列）想要达到限流的目标，又不会掐断流量，使得流量更加平滑？能够思考漏桶算法。 我为啥在漏桶算法这节加上 MQ 生产队列呢？因为在我的了解中，这种限流算法，就是 MQ 生产队列的利用办法。无论生产音讯的频率如何，MQ的消费者的生产频率下限是固定的。 有差异吗？有。漏桶算法中定义的是“桶容量固定。当水超过桶的容量时，会被溢出抛弃”。而 MQ 的惯例用法是“削峰填谷”，音讯能够在队列中积压，而后满满生产，但不会轻易抛弃。其实这也合乎通常的理论利用场景。真要实现漏桶算法的要求也行，齐全给队列设置为固定长度。 ...

引言高并发的零碎通常有三把利器：缓存、降级和限流。 缓存：缓存是进步零碎访问速度，缓解CPU解决压力的要害，同时能够进步零碎的解决容量。降级：降级是在忽然的压力剧增的状况，依据业务以及流量对一些服务和页面的策略降级，以此开释服务器资源。限流：限流是对于并发拜访/申请进行限速，或者一个工夫窗口内限速爱护零碎，一旦达到限度速度能够拒绝服务、排队或者期待。 限流算法令牌桶和和漏桶，比方Google的Guava的RateLimiter进行令牌痛管制。 漏桶算法漏桶算法是把流量比作水，水先放在桶外面并且以限定的速度出水，水过多会间接溢出，就会拒绝服务。 破绽存在出水口、进水口，出水口以肯定速率出水，并且有最大出水率。 在漏斗没有水的时候： 进水的速率小于等于最大出水率，那么出水速率等于进水速率，此时不会积水。如果进水速率大于最大出水速率，那么，漏斗以最大速率出水，此时，多余的水会积在漏斗中。如果漏斗有水的时候： 出水为最大速率。如果漏斗未满并且有进水，那么这些水会积在漏斗。如果漏斗已满并且有进水，那么水会溢出到漏斗外。令牌桶算法对于很多利用场景来说，除了要求可能限度数据的均匀传输速率外，还要求容许某种程度的突发传输。这个时候应用令牌桶算法比拟适合。 令牌桶算法以恒定的速率产生令牌，之后再把令牌放回到桶当中，令牌桶有一个容量，当令牌桶满了的时候，再向其中放令牌会被间接抛弃， RateLimiter 用法https://github.com/google/guava 首先增加Maven依赖： <!-- https://mvnrepository.com/artifact/com.google.guava/guava --> <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>31.1-jre</version> </dependency>acquire(int permits) 函数次要用于获取 permits 个令牌，并计算须要期待多长时间，进而挂起期待，并将该值返回。 import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.RateLimiter; import java.text.SimpleDateFormat; import java.util.Date; import java.util.concurrent.Executors;public class Test { public static void main(String[] args) { ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(100)); // 指定每秒放1个令牌 RateLimiter limiter = RateLimiter.create(1); for (int i = 1; i < 50; i++) { // 申请RateLimiter, 超过permits会被阻塞 //acquire(int permits)函数次要用于获取permits个令牌，并计算须要期待多长时间，进而挂起期待，并将该值返回 Double acquire = null; if (i == 1) { // `acquire 1` 时，并没有任何期待 0.0 秒 间接预生产了1个令牌 acquire = limiter.acquire(1); } else if (i == 2) { // `acquire 10`时，因为之前预生产了 1 个令牌，故而期待了1秒，之后又预生产了10个令牌 acquire = limiter.acquire(10); } else if (i == 3) { // `acquire 2` 时，因为之前预生产了 10 个令牌，故而期待了10秒，之后又预生产了2个令牌 acquire = limiter.acquire(2); } else if (i == 4) { //`acquire 20` 时，因为之前预生产了 2 个令牌，故而期待了2秒，之后又预生产了20个令牌 acquire = limiter.acquire(20); } else { // `acquire 2` 时，因为之前预生产了 2 个令牌，故而期待了2秒，之后又预生产了2个令牌 acquire = limiter.acquire(2); } executorService.submit(new Task("获取令牌胜利，获取耗：" + acquire + " 第 " + i + " 个工作执行")); } }}class Task implements Runnable { String str; public Task(String str) { this.str = str; } @Override public void run() { SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"); System.out.println(sdf.format(new Date()) + " | " + Thread.currentThread().getName() + str); }}一个RateLimiter次要定义了发放permits的速率。如果没有额定的配置，permits将以固定的速度调配，单位是每秒多少permits。默认状况下，Permits将会被稳固的平缓的发放。 ...

封面图 嗨~大家好啊，我是阿壮，一个有情怀的程序员。面试中常常会问到限流的办法有哪些，我整顿了常见的限流算法如下 固定窗口限流算法滑动窗口限流算法漏桶算法令牌桶算法限流是什么？限流顾名思义就是限流流量，也叫流量管制，在零碎面临高并发、大流量申请的状况下，限度新的流量对系统的拜访，从而保证系统服务的安全性。举个例子，一个电影院只有 200 个座位，并发售 200 张门票，每个门票代表一个人，代表这个电影院只能够进入 200 人，多余的人就买不到票也就进不了电影院。 为什么要实现限流限流起到保证系统稳固的作用。如果一个零碎不限度流量，如果有大量的流量指向零碎，比方在秒杀流动、双十一促销等场景下，点击量剧增，用户的流量巨增，然而服务的解决能力是无限的，如果不能解决好大流量申请，零碎很容易产生瘫痪。 固定窗口限流算法固定窗口限流算法又称计数器算法（Fixed Window），通过在单位工夫内保护的计数器来管制该工夫单位内的最大访问量。 设置没小时不超过 3 个申请，超出阈值则拒绝请求，未超出阈值则承受申请并将计数器加 1，当工夫窗口完结时，重置计数器为 0。 长处实现简略，容易了解。 毛病一段时间内服务不可用。比方 10s 内限流 100 此申请，然而 0s-1s 就来了 100 个申请，则 2s-10s 的申请都是被回绝的。 窗口切换时可能会呈现两倍阈值的申请。比方 10s 内限流 100 此申请。0s-9.99s 没有申请，而在 9.99s-10s 之间来了 100 个申请，当切换到下一个窗口时，在 0s-0.01s 之间来了 100 个申请，而咱们的阈值设置的是每 10s100 次申请，通过的申请达到了阈值的两倍 滑动窗口限流算法滑动窗口限流算法是固定窗口限流算法的改良，补救了固定窗口限流算法可能会呈现两倍阈值申请的毛病。为了避免刹时流量，能够把固定窗口近一步划分成多个格子，每次向后挪动一小格，而不是固定窗口大小。 在滑动窗口的状况加就不存在窗口切换导致申请达到阈值两倍的状况，因为它基本不会切换窗口，每次都是往后挪动一格，如图所示，假如 1s 内只承受 100 个申请，在绿色地位承受了 100 个申请，当到了黄色地位，应为在窗口内绿色曾经达到了 100 个申请的阈值，所以在黄色地位就不会再承受申请。 特点补救了固定窗口限流算法的毛病即防止了固定窗口算法在窗口切换时可能会产生两倍于阈值流量申请的问题。和漏桶算法相比，新来的申请也可能被解决到，防止了漏桶算法的饥饿问题。漏桶算法 举个例子，如果有一个桶，桶上面有一个小洞，每秒以固定的速率在滴水，当注入的水超过桶的容量时就会溢出。这里的滴水就代表咱们解决的申请，注水就代表用户流量，桶就代表咱们存储的申请，溢出就代表被抛弃的申请。在零碎看来，申请永远是以平滑的传输速率过去，从而起到了爱护零碎的作用。漏桶算法实现能够借助队列来实现，队列中存储咱们须要解决的申请即可。 特点漏桶的漏出速率是固定的，能够起到整流的作用。不能解决流量突发的问题。假如桶的容量是 5，“漏”的速度是 2 个/s，当忽然有 10 个申请时，首先抛弃 5 一个申请，还有 5 个申请放到桶中，而这 5 个申请还只是放在桶中在期待解决，所以并没有解决流量突发的问题。 ...

解决问题对象: 并发流量 解决方案：1、计数器：2、滑动窗口计数器：redis，zset数据结构，权重score存工夫戳，应用rangeByscore查问指定时间段内的3、漏桶：4、令牌桶： 方才去查了一些材料 如同redis根本都间接提供了 或者说构造能够实现 这几个限流 简略计数器：hash 存timestamp 和 count，每次进去在间隔timestamp有余指定时差timeLimit时，判断count是否大于countLimit，大于阻止拜访，否则累加count；超过timeLimit时，重置timestamp 和 count窗口计数器：zset value和score都存timestamp，每次申请先移除窗口期timeLimit之前的记录，对窗口期内的记录总数count与countLimit做比拟，如果超出了，阻止拜访 漏桶：Redis 4.0 提供了一个限流 Redis 模块，名称为 redis-cell，该模块提供漏斗算法，并提供原子的限流指令 这个略微有点不好了解，我要再看看令牌桶：list当做桶，length当做令牌数，或者初始化一个一般key-value，value存令牌数，如果length和value <= 0 则阻止拜访，否则pop或decr  另开一个常驻工作  按固定速率push或者incr  length或value=令牌下限则进行加牌，用while+sleep来实现这个加牌过程 漏桶：视用户的申请为水，发动申请是向桶内注水，解决申请是出水。这个过程跟令牌桶相同。 令牌桶是初始化为满桶令牌，漏桶是初始化为空桶水 令牌桶【资格】是空桶则阻止申请，漏桶【负载】是满桶阻止申请 令牌桶的加桶速率是  每 1/qps 秒+1，取牌速率跟申请同步； 漏桶的出水速率是qps，注水速率跟申请同步 令牌桶另开工作加牌（加资格），漏桶另开工作出水（消化申请） 这里好奇初始化桶的容量该是多少 QPS？ QPS = 并发量/均匀响应时长答案：并发量

大家好，我是 yes。 明天来说说限流的相干内容，包含常见的限流算法、单机限流场景、分布式限流场景以及一些常见限流组件。 当然在介绍限流算法和具体场景之前咱们先得明确什么是限流，为什么要限流？。 任何技术都要搞清它的起源，技术的产生来自痛点，明确痛点咱们能力抓住要害隔靴搔痒。 限流是什么？首先来解释下什么是限流？ 在日常生活中限流很常见，例如去有些景区玩，每天售卖的门票数是无限的，例如 2000 张，即每天最多只有 2000 集体能进去玩耍。 题外话：我之前看到个新闻，最不想卖门票的景区“卢旺达火山公园”，每天就卖 32 张，并且每张门票须要 1 万元！ 再回到主题，那在咱们工程下限流是什么呢？限度的是 「流」，在不同场景下「流」的定义不同，能够是每秒申请数、每秒事务处理数、网络流量等等。 而通常咱们说的限流指代的是 限度达到零碎的并发申请数，使得零碎可能失常的解决 局部 用户的申请，来保证系统的稳定性。 限流不可避免的会造成用户的申请变慢或者被拒的状况，从而会影响用户体验。因而限流是须要在用户体验和零碎稳定性之间做均衡的，即咱们常说的 trade off。 对了，限流也称流控（流量管制）。 为什么要限流？后面咱们有提到限流是为了保证系统的稳定性。 日常的业务上有相似秒杀流动、双十一大促或者突发新闻等场景，用户的流量突增，后端服务的解决能力是无限的，如果不能解决好突发流量，后端服务很容易就被打垮。 亦或是爬虫等不失常流量，咱们对外裸露的服务都要以最大歹意去防范咱们的调用者。咱们不分明调用者会如何调用咱们的服务。假如某个调用者开几十个线程一天二十四小时疯狂调用你的服务，不做啥解决咱服务也算完了。更胜的还有DDos攻打。 还有对于很多第三方开发平台来说，不仅仅是为了防范不失常流量，也是为了资源的偏心利用，有些接口都收费给你用了，资源都不可能始终都被你占着吧，他人也得调的。 当然加钱的话好磋商。 在之前公司还做过一个零碎，过后SaaS版本还没进去。因而零碎须要部署到客户方。 过后老板的要求是，咱们须要给他个限流降级版本，岂但零碎出的计划是降级后的计划，外围接口每天最多只能调用20次，还须要限度零碎所在服务器的配置和数量，即限度部署的服务器的CPU核数等，还限度所有部署的服务器数量，避免客户集群部署，进步零碎的性能。 当然这所有须要能动静配置，因为加钱的话好磋商。客户始终都不晓得。 预计老板在等客户说感觉零碎有点慢吧。而后就搞个2.0版本？我让咱们研发部加班加点给你搞进去。 小结一下，限流的实质是因为后端解决能力无限，须要截掉超过解决能力之外的申请，亦或是为了平衡客户端对服务端资源的偏心调用，避免一些客户端饿死。 常见的限流算法无关限流算法我给出了对应的图解和相干伪代码，有些人喜爱看图，有些人更喜爱看代码。 计数限流最简略的限流算法就是计数限流了，例如零碎能同时解决100个申请，保留一个计数器，解决了一个申请，计数器加一，一个申请处理完毕之后计数器减一。 每次申请来的时候看看计数器的值，如果超过阈值要么回绝。 十分的简略粗犷，计数器的值要是存内存中就算单机限流算法。存核心存储里，例如 Redis 中，集群机器拜访就算分布式限流算法。 长处就是：简略粗犷，单机在 Java 中可用 Atomic 等原子类、分布式就 Redis incr。 毛病就是：假如咱们容许的阈值是1万，此时计数器的值为0， 当1万个申请在前1秒内一股脑儿的都涌进来，这突发的流量可是顶不住的。缓缓的减少解决和一下子涌入对于程序来说是不一样的。 而且个别的限流都是为了限度在指定工夫距离内的访问量，因而还有个算法叫固定窗口。 固定窗口限流它相比于计数限流次要是多了个工夫窗口的概念。计数器每过一个工夫窗口就重置。规定如下： 申请次数小于阈值，容许拜访并且计数器 +1；申请次数大于阈值，回绝拜访；这个工夫窗口过了之后，计数器清零； 看起来如同很完满，实际上还是有缺点的。 固定窗口临界问题假如零碎每秒容许 100 个申请，假如第一个工夫窗口是 0-1s，在第 0.55s 处一下次涌入 100 个申请，过了 1 秒的工夫窗口后计数清零，此时在 1.05 s 的时候又一下次涌入100个申请。 ...

LastModified: 2019年6月14日10:37:39 主要是依靠 redis + lua 来实现限流器, 使用 lua 的原因是将多条命令合并在一起作为一个原子操作, 无需过多考虑并发. 计数器模式原理计数器算法是指在一段窗口时间内允许通过的固定数量的请求, 比如10次/秒, 500次/30秒. 如果设置的时间粒度越细, 那么限流会更平滑. 实现所使用的 Lua 脚本 -- 计数器限流-- 此处支持的最小单位时间是秒, 若将 expire 改成 pexpire 则可支持毫秒粒度.-- KEYS[1] string 限流的key-- ARGV[1] int 限流数-- ARGV[2] int 单位时间(秒)local cnt = tonumber(redis.call("incr", KEYS[1]))if (cnt == 1) then -- cnt 值为1说明之前不存在该值, 因此需要设置其过期时间 redis.call("expire", KEYS[1], tonumber(ARGV[2]))elseif (cnt > tonumber(ARGV[1])) then return -1end return cnt返回 -1 表示超过限流, 否则返回当前单位时间已通过的请求数key 可以但不限于以下的情况 ip + 接口user_id + 接口优点 ...

应用范围为服务访问量突然剧增，原因可能有多种外部的调用或内部的一些问题导致消息积压，对服务的访问超过服务所能处理的最大峰值，导致系统超时负载从而崩溃。业务场景举一些我们平常生活中的消费场景，例如：火车票、机票、门票等，通常来说这些服务在下单之后，后续的出票结果都是异步通知的，如果服务本身只支持每秒1000访问量，由于外部服务的原因突然访问量增加到每秒2000并发，这个时候服务接收者因为流量的剧增，超过了自己系统本身所能处理的最大峰值，如果没有对消息做限流措施，系统在这段时间内就会造成不可用，在生产环境这是一个很严重的问题，实际应用场景不止于这些，本文通过RabbitMQ来讲解如果对消费端做限流措施。 消费端限流机制RabbitMQ提供了服务质量保证 (QOS) 功能，对channel（通道）预先设置一定的消息数目，每次发送的消息条数都是基于预先设置的数目，如果消费端一旦有未确认的消息，这时服务端将不会再发送新的消费消息，直到消费端将消息进行完全确认，注意：此时消费端不能设置自动签收，否则会无效。 在 RabbitMQ v3.3.0 之后，放宽了限制，除了对channel设置之外，还可以对每个消费者进行设置。 以下为 Node.js 开发语言 amqplib 库对于限流实现提供的接口方法 prefetch export interface Channel extends events.EventEmitter { prefetch(count: number, global?: boolean): Promise<Replies.Empty>; ...}prefetch 参数说明： number：每次推送给消费端 N 条消息数目，如果这 N 条消息没有被ack，生产端将不会再次推送直到这 N 条消息被消费。global：在哪个级别上做限制，ture 为 channel 上做限制，false 为消费端上做限制，默认为 false。建立生产端生产端没什么变化，和正常声明一样，关于源码参见rabbitmq-prefetch（Node.js客户端版Demo） const amqp = require('amqplib');async function producer() { // 1. 创建链接对象 const connection = await amqp.connect('amqp://localhost:5672'); // 2. 获取通道 const channel = await connection.createChannel(); // 3. 声明参数 const exchangeName = 'qosEx'; const routingKey = 'qos.test001'; const msg = 'Producer：'; // 4. 声明交换机 await channel.assertExchange(exchangeName, 'topic', { durable: true }); for (let i=0; i<5; i++) { // 5. 发送消息 await channel.publish(exchangeName, routingKey, Buffer.from(`${msg} 第${i}条消息`)); } await channel.close();}producer();建立消费端const amqp = require('amqplib');async function consumer() { // 1. 创建链接对象 const connection = await amqp.connect('amqp://localhost:5672'); // 2. 获取通道 const channel = await connection.createChannel(); // 3. 声明参数 const exchangeName = 'qosEx'; const queueName = 'qosQueue'; const routingKey = 'qos.#'; // 4. 声明交换机、对列进行绑定 await channel.assertExchange(exchangeName, 'topic', { durable: true }); await channel.assertQueue(queueName); await channel.bindQueue(queueName, exchangeName, routingKey); // 5. 限流参数设置 await channel.prefetch(1, false); // 6. 限流，noAck参数必须设置为false await channel.consume(queueName, msg => { console.log('Consumer：', msg.content.toString()); // channel.ack(msg); }, { noAck: false });}consumer();未确认消息情况测试在 consumer 中我们暂且将 channel.ack(msg) 注释掉，分别启动生产者和消费者，看看是什么情况？ ...

Java单机限流可以使用AtomicInteger，RateLimiter或Semaphore来实现，但是上述方案都不支持集群限流。集群限流的应用场景有两个，一个是网关，常用的方案有Nginx限流和Spring Cloud Gateway，另一个场景是与外部或者下游服务接口的交互，因为接口限制必须进行限流。 本文的主要内容为：Redis和Lua的使用场景和注意事项，比如说KEY映射的问题。Spring Cloud Gateway中限流的实现。集群限流的难点 在上篇Guava RateLimiter的分析文章中，我们学习了令牌桶限流算法的原理，下面我们就探讨一下，如果将RateLimiter扩展，让它支持集群限流，会遇到哪些问题。 RateLimiter会维护两个关键的参数nextFreeTicketMicros和storedPermits，它们分别是下一次填充时间和当前存储的令牌数。当RateLimiter的acquire函数被调用时，也就是有线程希望获取令牌时，RateLimiter会对比当前时间和nextFreeTicketMicros，根据二者差距，刷新storedPermits，然后再判断更新后的storedPermits是否足够，足够则直接返回，否则需要等待直到令牌足够(Guava RateLimiter的实现比较特殊，并不是当前获取令牌的线程等待，而是下一个获取令牌的线程等待)。 由于要支持集群限流，所以nextFreeTicketMicros和storedPermits这两个参数不能只存在JVM的内存中，必须有一个集中式存储的地方。而且，由于算法要先获取两个参数的值，计算后在更新两个数值，这里涉及到竞态限制，必须要处理并发问题。 集群限流由于会面对相比单机更大的流量冲击，所以一般不会进行线程等待，而是直接进行丢弃，因为如果让拿不到令牌的线程进行睡眠，会导致大量的线程堆积，线程持有的资源也不会释放，反而容易拖垮服务器。Redis和Lua 分布式限流本质上是一个集群并发问题，Redis单进程单线程的特性，天然可以解决分布式集群的并发问题。所以很多分布式限流都基于Redis，比如说Spring Cloud的网关组件Gateway。 Redis执行Lua脚本会以原子性方式进行，单线程的方式执行脚本，在执行脚本时不会再执行其他脚本或命令。并且，Redis只要开始执行Lua脚本，就会一直执行完该脚本再进行其他操作，所以Lua脚本中不能进行耗时操作。使用Lua脚本，还可以减少与Redis的交互，减少网络请求的次数。 Redis中使用Lua脚本的场景有很多，比如说分布式锁，限流，秒杀等，总结起来，下面两种情况下可以使用Lua脚本:使用 Lua 脚本实现原子性操作的CAS，避免不同客户端先读Redis数据，经过计算后再写数据造成的并发问题。前后多次请求的结果有依赖时，使用 Lua 脚本将多个请求整合为一个请求。 但是使用Lua脚本也有一些注意事项：要保证安全性，在 Lua 脚本中不要定义自己的全局变量，以免污染 Redis内嵌的Lua环境。因为Lua脚本中你会使用一些预制的全局变量，比如说redis.call()要注意 Lua 脚本的时间复杂度，Redis 的单线程同样会阻塞在 Lua 脚本的执行中。使用 Lua 脚本实现原子操作时，要注意如果 Lua 脚本报错，之前的命令无法回滚，这和Redis所谓的事务机制是相同的。一次发出多个 Redis 请求，但请求前后无依赖时，使用 pipeline，比 Lua 脚本方便。Redis要求单个Lua脚本操作的key必须在同一个Redis节点上。解决方案可以看下文对Gateway原理的解析。性能测试 Redis虽然以单进程单线程模型进行操作，但是它的性能却十分优秀。总结来说，主要是因为:绝大部分请求是纯粹的内存操作采用单线程,避免了不必要的上下文切换和竞争条件内部实现采用非阻塞IO和epoll，基于epoll自己实现的简单的事件框架。epoll中的读、写、关闭、连接都转化成了事件，然后利用epoll的多路复用特性，绝不在io上浪费一点时间。 所以，在集群限流时使用Redis和Lua的组合并不会引入过多的性能损耗。我们下面就简单的测试一下，顺便熟悉一下涉及的Redis命令。# test.lua脚本的内容local test = redis.call(“get”, “test”)local time = redis.call(“get”, “time”)redis.call(“setex”, “test”, 10, “xx”)redis.call(“setex”, “time”, 10, “xx”)return {test, time}# 将脚本导入redis，之后调用不需再传递脚本内容redis-cli -a 082203 script load “$(cat test.lua)““b978c97518ae7c1e30f246d920f8e3c321c76907”# 使用redis-benchmark和evalsha来执行lua脚本redis-benchmark -a 082203 -n 1000000 evalsha b978c97518ae7c1e30f246d920f8e3c321c76907 0 ======1000000 requests completed in 20.00 seconds50 parallel clients3 bytes payloadkeep alive: 193.54% <= 1 milliseconds99.90% <= 2 milliseconds99.97% <= 3 milliseconds99.98% <= 4 milliseconds99.99% <= 5 milliseconds100.00% <= 6 milliseconds100.00% <= 7 milliseconds100.00% <= 7 milliseconds49997.50 requests per second 通过上述简单的测试，我们可以发现本机情况下，使用Redis执行Lua脚本的性能极其优秀，一百万次执行，99.99%在5毫秒以下。 本来想找一下官方的性能数据，但是针对Redis + Lua的性能数据较少，只找到了几篇个人博客，感兴趣的同学可以去探索。这篇文章有Lua和zadd的性能比较(具体数据请看原文，链接缺失的话，请看文末)。以上lua脚本的性能大概是zadd的70%-80%，但是在可接受的范围内，在生产环境可以使用。负载大概是zadd的1.5-2倍，网络流量相差不大，IO是zadd的3倍，可能是开启了AOF，执行了三次操作。Spring Cloud Gateway的限流实现 Gateway是微服务架构Spring Cloud的网关组件，它基于Redis和Lua实现了令牌桶算法的限流功能，下面我们就来看一下它的原理和细节吧。 Gateway基于Filter模式，提供了限流过滤器RequestRateLimiterGatewayFilterFactory。只需在其配置文件中进行配置，就可以使用。具体的配置感兴趣的同学自行学习，我们直接来看它的实现。 RequestRateLimiterGatewayFilterFactory依赖RedisRateLimiter的isAllowed函数来判断一个请求是否要被限流抛弃。public Mono<Response> isAllowed(String routeId, String id) { //routeId是ip地址，id是使用KeyResolver获取的限流维度id，比如说基于uri,IP或者用户等等。 Config routeConfig = loadConfiguration(routeId); // 每秒能够通过的请求数 int replenishRate = routeConfig.getReplenishRate(); // 最大流量 int burstCapacity = routeConfig.getBurstCapacity(); try { // 组装Lua脚本的KEY List<String> keys = getKeys(id); // 组装Lua脚本需要的参数，1是指一次获取一个令牌 List<String> scriptArgs = Arrays.asList(replenishRate + “”, burstCapacity + “”, Instant.now().getEpochSecond() + “”, “1”); // 调用Redis，tokens_left = redis.eval(SCRIPT, keys, args) Flux<List<Long>> flux = this.redisTemplate.execute(this.script, keys, scriptArgs); ….. // 省略 }static List<String> getKeys(String id) { String prefix = “request_rate_limiter.{” + id; String tokenKey = prefix + “}.tokens”; String timestampKey = prefix + “}.timestamp”; return Arrays.asList(tokenKey, timestampKey);} 需要注意的是getKeys函数的prefix包含了”{id}"，这是为了解决Redis集群键值映射问题。Redis的KeySlot算法中，如果key包含{}，就会使用第一个{}内部的字符串作为hash key，这样就可以保证拥有同样{}内部字符串的key就会拥有相同slot。Redis要求单个Lua脚本操作的key必须在同一个节点上，但是Cluster会将数据自动分布到不同的节点，使用这种方法就解决了上述的问题。 然后我们来看一下Lua脚本的实现，该脚本就在Gateway项目的resource文件夹下。它就是如同Guava的RateLimiter一样，实现了令牌桶算法，只不过不在需要进行线程休眠，而是直接返回是否能够获取。local tokens_key = KEYS[1] – request_rate_limiter.${id}.tokens 令牌桶剩余令牌数的KEY值local timestamp_key = KEYS[2] – 令牌桶最后填充令牌时间的KEY值local rate = tonumber(ARGV[1]) – replenishRate 令令牌桶填充平均速率local capacity = tonumber(ARGV[2]) – burstCapacity 令牌桶上限local now = tonumber(ARGV[3]) – 得到从 1970-01-01 00:00:00 开始的秒数local requested = tonumber(ARGV[4]) – 消耗令牌数量，默认 1 local fill_time = capacity/rate – 计算令牌桶填充满令牌需要多久时间local ttl = math.floor(fill_time*2) – 2 保证时间充足local last_tokens = tonumber(redis.call(“get”, tokens_key)) – 获得令牌桶剩余令牌数if last_tokens == nil then – 第一次时，没有数值，所以桶时满的 last_tokens = capacityendlocal last_refreshed = tonumber(redis.call(“get”, timestamp_key)) – 令牌桶最后填充令牌时间if last_refreshed == nil then last_refreshed = 0endlocal delta = math.max(0, now-last_refreshed) – 获取距离上一次刷新的时间间隔local filled_tokens = math.min(capacity, last_tokens+(deltarate)) – 填充令牌，计算新的令牌桶剩余令牌数 填充不超过令牌桶令牌上限。local allowed = filled_tokens >= requested local new_tokens = filled_tokenslocal allowed_num = 0if allowed then– 若成功，令牌桶剩余令牌数(new_tokens) 减消耗令牌数( requested )，并设置获取成功( allowed_num = 1 ) 。 new_tokens = filled_tokens - requested allowed_num = 1end – 设置令牌桶剩余令牌数( new_tokens ) ，令牌桶最后填充令牌时间(now) ttl是超时时间？redis.call(“setex”, tokens_key, ttl, new_tokens)redis.call(“setex”, timestamp_key, ttl, now)– 返回数组结果return { allowed_num, new_tokens }后记 Redis的主从异步复制机制可能丢失数据，出现限流流量计算不准确的情况，当然限流毕竟不同于分布式锁这种场景，对于结果的精确性要求不是很高，即使多流入一些流量，也不会影响太大。 正如Martin在他质疑Redis分布式锁RedLock文章中说的，Redis的数据丢弃了也无所谓时再使用Redis存储数据。I think it’s a good fit in situations where you want to share some transient, approximate, fast-changing data between servers, and where it’s not a big deal if you occasionally lose that data for whatever reason 接下来我们回来学习阿里开源的分布式限流组件sentinel，希望大家持续关注。 个人博客: Remcarpediem参考https://www.cnblogs.com/itren…压测的文章：https://www.fuwuqizhijia.com/…https://blog.csdn.net/forezp/...https://blog.csdn.net/xixingz...Matin RedLock http://martin.kleppmann.com/2… ...

本博客是深入研究Envoy Proxy和Istio.io 以及它如何实现更优雅的方式来连接和管理微服务系列文章的一部分。这是接下来几个部分的想法（将在发布时更新链接):断路器（第一部分）重试/超时（第二部分）分布式跟踪（第三部分）Prometheus的指标收集（第四部分）rate limiter（第五部分）第五部分 - rate limiterEnvoy ratelimit filtersEnvoy通过两个过滤器与Ratelimit服务集成：Network Level Filter： envoy为安装过滤器的侦听器上的每个新连接调用Ratelimit服务。这样，您可以对通过侦听器的每秒连接进行速率限制。HTTP Level Filter：Envoy为安装过滤器的侦听器上的每个新请求调用Ratelimit服务，路由表指定应调用Ratelimit服务。许多工作都在扩展HTTP过滤器的功能。envoy 配置 启用 http rate limiterhttp rate limiter 当请求的路由或虚拟主机具有与过滤器阶段设置匹配的一个或多个速率限制配置时，HTTP速率限制过滤器将调用速率限制服务。该路由可以选择包括虚拟主机速率限制配置。多个配置可以应用于请求。每个配置都会导致将描述符发送到速率限制服务。如果调用速率限制服务，并且任何描述符的响应超出限制，则返回429响应。速率限制过滤器还设置x-envoy-ratelimited标头。果在呼叫速率限制服务中出现错误或速率限制服务返回错误并且failure_mode_deny设置为true，则返回500响应。全部的配置如下： envoy.yaml: |- static_resources: listeners: - address: socket_address: address: 0.0.0.0 port_value: 8000 filter_chains: - filters: - name: envoy.http_connection_manager config: codec_type: auto stat_prefix: ingress_http access_log: - name: envoy.file_access_log config: path: “/dev/stdout” format: “[ACCESS_LOG][%START_TIME%] "%REQ(:METHOD)% %REQ(X-ENVOY-ORIGINAL-PATH?:PATH)% %PROTOCOL%" %RESPONSE_CODE% %RESPONSE_FLAGS% %BYTES_RECEIVED% %BYTES_SENT% %DURATION% %RESP(X-ENVOY-UPSTREAM-SERVICE-TIME)% "%REQ(X-FORWARDED-FOR)%" "%REQ(USER-AGENT)%" "%REQ(X-REQUEST-ID)%" "%REQ(:AUTHORITY)%" "%UPSTREAM_HOST%" "%DOWNSTREAM_REMOTE_ADDRESS_WITHOUT_PORT%"\n” route_config: name: local_route virtual_hosts: - name: gateway domains: - “*” routes: - match: prefix: “/cost” route: cluster: cost rate_limits: # enable rate limit checks for the greeter service actions: - destination_cluster: {} http_filters: - name: envoy.rate_limit # enable the Rate Limit filter config: domain: envoy - name: envoy.router config: {} clusters: - name: cost connect_timeout: 0.25s type: strict_dns lb_policy: round_robin hosts: - socket_address: address: cost.sgt port_value: 80 - name: rate_limit_cluster type: strict_dns connect_timeout: 0.25s lb_policy: round_robin http2_protocol_options: {} hosts: - socket_address: address: limiter.sgt port_value: 80 rate_limit_service: grpc_service: envoy_grpc: cluster_name: rate_limit_cluster timeout: 0.25s admin: access_log_path: “/dev/null” address: socket_address: address: 0.0.0.0 port_value: 9000 通过配置文件可以看出，本demo设置的是一个全局的http filter rate limiter。尽管分布式熔断通常在控制分布式系统中的吞吐量方面非常有效，但有时它不是非常有效并且需要全局速率限制。最常见的情况是当大量主机转发到少量主机并且平均请求延迟较低时（例如，对数据库服务器的连接/请求）。如果目标主机已备份，则下游主机将淹没上游群集。在这种情况下，在每个下游主机上配置足够严格的断路限制是非常困难的，这样系统在典型的请求模式期间将正常运行，但在系统开始出现故障时仍能防止级联故障。全局速率限制是这种情况的一个很好的解决方案。编写rate limiter 服务Envoy直接通过gRPC与速率限制服务集成。Envoy要求速率限制服务支持rls.proto中指定的gRPC IDL。有关API如何工作的更多信息，请参阅IDL文档。本身envoy 只是提供了限流的接口，没有具体的实现，所以必须自己实现一个限流器。下面只是简单实现一下，给大家一个思路。具体的代码如下：package mainimport ( “log” “net” “time” rls “github.com/envoyproxy/go-control-plane/envoy/service/ratelimit/v2” “github.com/juju/ratelimit” “golang.org/x/net/context” “google.golang.org/grpc” “google.golang.org/grpc/reflection”)// server is used to implement rls.RateLimitServicetype server struct { bucket *ratelimit.Bucket}func (s *server) ShouldRateLimit(ctx context.Context, request *rls.RateLimitRequest) (rls.RateLimitResponse, error) { // logic to rate limit every second request var overallCode rls.RateLimitResponse_Code if s.bucket.TakeAvailable(1) == 0 { overallCode = rls.RateLimitResponse_OVER_LIMIT } else { overallCode = rls.RateLimitResponse_OK } response := &rls.RateLimitResponse{OverallCode: overallCode} return response, nil}func main() { // create a TCP listener on port 8089 lis, err := net.Listen(“tcp”, “:8089”) if err != nil { log.Fatalf(“failed to listen: %v”, err) } log.Printf(“listening on %s”, lis.Addr()) // create a gRPC server and register the RateLimitService server s := grpc.NewServer() rls.RegisterRateLimitServiceServer(s, &server{ bucket: ratelimit.NewBucket(100time.Microsecond, 100), }) reflection.Register(s) if err := s.Serve(lis); err != nil { log.Fatalf(“failed to serve: %v”, err) }}具体项目，查阅github。PS:使用了令牌桶算法来限流。令牌桶算法(Token Bucket)和 Leaky Bucket 效果一样但方向相反的算法,更加容易理解.随着时间流逝,系统会按恒定1/QPS时间间隔(如果QPS=100,则间隔是10ms)往桶里加入Token(想象和漏洞漏水相反,有个水龙头在不断的加水),如果桶已经满了就不再加了.新请求来临时,会各自拿走一个Token,如果没有Token可拿了就阻塞或者拒绝服务.该实现存在单点风险。Dockerfile均在代码仓库中，大家可以构建镜像自己测试。结论本文简单讲了envoy的 rate limit功能，提供了全局限流的配置文件，并简单实现了一个基于令牌桶的限流器。希望能帮助你理解Envoy的限速过滤器如何跟gRPC协议协同工作。 ...

限流每个API接口都是有访问上限的，当访问频率或者并发量超过其承受范围时候，我们就必须考虑限流来保证接口的可用性或者降级可用性。即接口也需要安装上保险丝,以防止非预期的请求对系统压力过大而引起的系统瘫痪。通常的策略就是拒绝多余的访问，或者让多余的访问排队等待服务，或者引流。如果要准确的控制QPS，简单的做法是维护一个单位时间内的Counter，如判断单位时间已经过去，则将Counter重置零。此做法被认为没有很好的处理单位时间的边界，比如在前一秒的最后一毫秒里和下一秒的第一毫秒都触发了最大的请求数，将目光移动一下，就看到在两毫秒内发生了两倍的QPS。限流算法常用的更平滑的限流算法有两种：漏桶算法和令牌桶算法。漏桶算法漏桶(Leaky Bucket)算法思路很简单，水(请求)先进入到漏桶里，漏桶以一定的速度出水(接口有响应速率)，当水流入速度过大会直接溢出(访问频率超过接口响应速率)，然后就拒绝请求，可以看出漏桶算法能强行限制数据的传输速率。示意图如下:可见这里有两个变量，一个是桶的大小，支持流量突发增多时可以存多少的水(burst)，另一个是水桶漏洞的大小(rate)。因为漏桶的漏出速率是固定的参数，所以，即使网络中不存在资源冲突(没有发生拥塞)，漏桶算法也不能使流突发(burst)到端口速率。因此，漏桶算法对于存在突发特性的流量来说缺乏效率。令牌桶算法令牌桶算法(Token Bucket)和 Leaky Bucket 效果一样但方向相反的算法，更加容易理解。随着时间流逝，系统会按恒定1/QPS时间间隔(如果QPS=100,则间隔是10ms)往桶里加入Token(想象和漏洞漏水相反，有个水龙头在不断的加水)，如果桶已经满了就不再加了。新请求来临时，会各自拿走一个Token，如果没有Token可拿了就阻塞或者拒绝服务。令牌桶的另外一个好处是可以方便的改变速度。一旦需要提高速率，则按需提高放入桶中的令牌的速率。一般会定时(比如100毫秒)往桶中增加一定数量的令牌，有些变种算法则实时的计算应该增加的令牌的数量。

问题spring-cloud-gateway 网关新增了一个限流功能，使用的是模块自带的限流过滤器 RequestRateLimiterGatewayFilterFactory，基于令牌桶算法，通过 redis 实现。其原理是 redis 中针对每个限流要素（比如针对接口限流），保存 2 个 key：tokenKey（令牌数量），timeKey（调用时间）。每次接口调用时，更新 tokenKey 的值为：原先的值 + （当前时间 - 原先时间）* 加入令牌的速度，如果新的 tokenKey 的值大于 1，那么允许调用，否则不允许；同时更新 redis 中 tokenKey，timeKey 的值。整个过程通过 lua 脚本实现。在加入限流功能之前，500 客户端并发访问，tps 为 6800 req/s，50% 时延为 70ms；加入限流功能之后，tps 为 2300 req/s，50% 时延为 205ms，同时，原先 cpu 占用率几乎 600%（6 核） 变成不到 400%（cpu 跑不满了）。2. 排查和解决过程2.1 单个 CPU 跑满查看单个线程的 cpu 占用：[root@auth-service imf2]# top -Hp 29360top - 15:16:27 up 102 days, 18:04, 1 user, load average: 1.61, 0.72, 0.34Threads: 122 total, 9 running, 113 sleeping, 0 stopped, 0 zombie%Cpu(s): 42.0 us, 7.0 sy, 0.0 ni, 49.0 id, 0.0 wa, 0.0 hi, 2.0 si, 0.0 stKiB Mem : 7678384 total, 126844 free, 3426148 used, 4125392 buff/cacheKiB Swap: 6291452 total, 2212552 free, 4078900 used. 3347956 avail Mem PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND29415 root 20 0 6964708 1.1g 14216 R 97.9 15.1 3:01.65 java29392 root 20 0 6964708 1.1g 14216 R 27.0 15.1 0:45.42 java29391 root 20 0 6964708 1.1g 14216 R 24.8 15.1 0:43.95 java29387 root 20 0 6964708 1.1g 14216 R 23.8 15.1 0:46.38 java29388 root 20 0 6964708 1.1g 14216 R 23.4 15.1 0:48.21 java29390 root 20 0 6964708 1.1g 14216 R 23.0 15.1 0:45.93 java29389 root 20 0 6964708 1.1g 14216 R 22.3 15.1 0:44.36 java线程 29415 几乎跑满了 cpu，查看是什么线程：[root@auth-service imf2]# printf ‘%x\n’ 2941572e7[root@auth-service imf2]# jstack 29360 | grep 72e7"lettuce-nioEventLoop-4-1" #40 daemon prio=5 os_prio=0 tid=0x00007f604cc92000 nid=0x72e7 runnable [0x00007f606ce90000]果然是操作 redis 的线程，和预期一致。查看 redis：cpu 占用率不超过 15%，没有 10ms 以上的慢查询。应该不会是 redis 的问题。查看线程栈信息：通过以下脚本每秒记录一次 jstack：[root@eureka2 jstack]# cat jstack.sh#!/bin/shi=0while [ $i -lt 30 ]; do/bin/sleep 1i=expr $i + 1jstack 29360 > “$i”.txtdone查看 lettuce 线程主要执行哪些函数：“lettuce-nioEventLoop-4-1” #36 daemon prio=5 os_prio=0 tid=0x00007f1eb07ab800 nid=0x4476 runnable [0x00007f1eec8fb000] java.lang.Thread.State: RUNNABLE at sun.misc.URLClassPath$Loader.findResource(URLClassPath.java:715) at sun.misc.URLClassPath.findResource(URLClassPath.java:215) at java.net.URLClassLoader$2.run(URLClassLoader.java:569) at java.net.URLClassLoader$2.run(URLClassLoader.java:567) at java.security.AccessController.doPrivileged(Native Method) at java.net.URLClassLoader.findResource(URLClassLoader.java:566) at org.springframework.boot.loader.LaunchedURLClassLoader.findResource(LaunchedURLClassLoader.java:57) at java.lang.ClassLoader.getResource(ClassLoader.java:1096) at org.springframework.core.io.ClassPathResource.resolveURL(ClassPathResource.java:155) at org.springframework.core.io.ClassPathResource.getURL(ClassPathResource.java:193) at org.springframework.core.io.AbstractFileResolvingResource.lastModified(AbstractFileResolvingResource.java:220) at org.springframework.scripting.support.ResourceScriptSource.retrieveLastModifiedTime(ResourceScriptSource.java:119) at org.springframework.scripting.support.ResourceScriptSource.isModified(ResourceScriptSource.java:109) - locked <0x000000008c074d00> (a java.lang.Object) at org.springframework.data.redis.core.script.DefaultRedisScript.getSha1(DefaultRedisScript.java:89) - locked <0x000000008c074c10> (a java.lang.Object) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor.eval(DefaultReactiveScriptExecutor.java:113) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor.lambda$execute$0(DefaultReactiveScriptExecutor.java:105) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor$$Lambda$1268/1889039573.doInRedis(Unknown Source) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor.lambda$execute$6(DefaultReactiveScriptExecutor.java:167) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor$$Lambda$1269/1954779522.get(Unknown Source) at reactor.core.publisher.FluxDefer.subscribe(FluxDefer.java:46)可知该线程主要在执行 ReactiveRedisTemplate 类的 execute(RedisScript<T> script, List<K> keys, List<?> args) 方法，即运行 lua 脚本。猜想：既然是因为 lettuce-nioEventLoop 线程跑满了 cpu，那么通过创建多个 lettuce-nioEventLoop 线程，以充分利用多核的特点，是否可以解决呢？以下为源码分析阶段：// 1. RedisConnectionFactory bean 的创建依赖 ClientResources@Bean@ConditionalOnMissingBean(RedisConnectionFactory.class)public LettuceConnectionFactory redisConnectionFactory( ClientResources clientResources) throws UnknownHostException { LettuceClientConfiguration clientConfig = getLettuceClientConfiguration( clientResources, this.properties.getLettuce().getPool()); return createLettuceConnectionFactory(clientConfig);}// 2. ClientResources bean 的创建如下@Bean(destroyMethod = “shutdown”)@ConditionalOnMissingBean(ClientResources.class)public DefaultClientResources lettuceClientResources() { return DefaultClientResources.create();}public static DefaultClientResources create() { return builder().build();}// 3. 创建 EventLoopGroupProvider 对象protected DefaultClientResources(Builder builder) { this.builder = builder; // 默认为 null，执行这块代码 if (builder.eventLoopGroupProvider == null) { // 设置处理 redis 连接的线程数：默认为 // Math.max(1, // SystemPropertyUtil.getInt(“io.netty.eventLoopThreads”, // Math.max(MIN_IO_THREADS, Runtime.getRuntime().availableProcessors()))); // 针对多核处理器，该值一般等于 cpu 的核的数量 int ioThreadPoolSize = builder.ioThreadPoolSize; if (ioThreadPoolSize < MIN_IO_THREADS) { logger.info(“ioThreadPoolSize is less than {} ({}), setting to: {}”, MIN_IO_THREADS, ioThreadPoolSize, MIN_IO_THREADS); ioThreadPoolSize = MIN_IO_THREADS; } this.sharedEventLoopGroupProvider = false; // 创建 EventLoopGroupProvider 对象 this.eventLoopGroupProvider = new DefaultEventLoopGroupProvider(ioThreadPoolSize); } else { this.sharedEventLoopGroupProvider = true; this.eventLoopGroupProvider = builder.eventLoopGroupProvider; } // 以下代码省略 …}// 4. 通过 EventLoopGroupProvider 创建 EventExecutorGroup 对象public static <T extends EventExecutorGroup> EventExecutorGroup createEventLoopGroup(Class<T> type, int numberOfThreads) { if (DefaultEventExecutorGroup.class.equals(type)) { return new DefaultEventExecutorGroup(numberOfThreads, new DefaultThreadFactory(“lettuce-eventExecutorLoop”, true)); } // 我们采用的是 Nio 模式，会执行这个分支 if (NioEventLoopGroup.class.equals(type)) { return new NioEventLoopGroup(numberOfThreads, new DefaultThreadFactory(“lettuce-nioEventLoop”, true)); } if (EpollProvider.isAvailable() && EpollProvider.isEventLoopGroup(type)) { return EpollProvider.newEventLoopGroup(numberOfThreads, new DefaultThreadFactory(“lettuce-epollEventLoop”, true)); } if (KqueueProvider.isAvailable() && KqueueProvider.isEventLoopGroup(type)) { return KqueueProvider.newEventLoopGroup(numberOfThreads, new DefaultThreadFactory(“lettuce-kqueueEventLoop”, true)); } throw new IllegalArgumentException(String.format(“Type %s not supported”, type.getName()));}// 5. NioEventLoopGroup 继承了 MultithreadEventLoopGroup；// 创建了多个 NioEventLoop；// 每个 NioEventLoop 都是单线程；// 每个 NioEventLoop 都可以处理多个连接。public class NioEventLoopGroup extends MultithreadEventLoopGroup { … }public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup { … }public final class NioEventLoop extends SingleThreadEventLoop { … }以上分析可知，默认创建的 RedisConnectionFactory bean 其实是支持多线程的，但通过 jstack 等方式查看 lettuce-nioEventLoop 线程却只有一个。[root@ ~]# ss | grep 6379tcp ESTAB 0 0 ::ffff:10.201.0.27:36184 ::ffff:10.201.0.30:6379查看 redis 连接，发现只有一个。在 Netty 中，一个 EventLoop 线程可以处理多个 Channel，但是一个 Channel 只能绑定到一个 EventLoop，这是基于线程安全和同步考虑而设计的。这解释了为什么只有一个 lettuce-nioEventLoop。下面继续分析为什么会只有一个连接呢？继续源码分析：// 1. 创建 RedisConnectionFactory bean@Bean@ConditionalOnMissingBean(RedisConnectionFactory.class)public LettuceConnectionFactory redisConnectionFactory( ClientResources clientResources) throws UnknownHostException { LettuceClientConfiguration clientConfig = getLettuceClientConfiguration( clientResources, this.properties.getLettuce().getPool()); return createLettuceConnectionFactory(clientConfig);}// 2. 查看 createLettuceConnectionFactory(clientConfig) 方法private LettuceConnectionFactory createLettuceConnectionFactory( LettuceClientConfiguration clientConfiguration) { if (getSentinelConfig() != null) { return new LettuceConnectionFactory(getSentinelConfig(), clientConfiguration); } if (getClusterConfiguration() != null) { return new LettuceConnectionFactory(getClusterConfiguration(), clientConfiguration); } // 没有哨兵模式，没有集群，执行这块代码 return new LettuceConnectionFactory(getStandaloneConfig(), clientConfiguration);}// 3. 获取 redis 连接private boolean shareNativeConnection = true;public LettuceReactiveRedisConnection getReactiveConnection() { // 默认为 true return getShareNativeConnection() ? new LettuceReactiveRedisConnection(getSharedReactiveConnection(), reactiveConnectionProvider) : new LettuceReactiveRedisConnection(reactiveConnectionProvider);}LettuceReactiveRedisConnection(StatefulConnection<ByteBuffer, ByteBuffer> sharedConnection, LettuceConnectionProvider connectionProvider) { Assert.notNull(sharedConnection, “Shared StatefulConnection must not be null!”); Assert.notNull(connectionProvider, “LettuceConnectionProvider must not be null!”); this.dedicatedConnection = new AsyncConnect(connectionProvider, StatefulConnection.class); this.pubSubConnection = new AsyncConnect(connectionProvider, StatefulRedisPubSubConnection.class); // 包装 sharedConnection this.sharedConnection = Mono.just(sharedConnection);}protected Mono<? extends StatefulConnection<ByteBuffer, ByteBuffer>> getConnection() { // 直接返回 sharedConnection if (sharedConnection != null) { return sharedConnection; } return getDedicatedConnection();}// 4. shareNativeConnection 是怎么来的protected StatefulConnection<ByteBuffer, ByteBuffer> getSharedReactiveConnection() { return shareNativeConnection ? getOrCreateSharedReactiveConnection().getConnection() : null;}private SharedConnection<ByteBuffer> getOrCreateSharedReactiveConnection() { synchronized (this.connectionMonitor) { if (this.reactiveConnection == null) { this.reactiveConnection = new SharedConnection<>(reactiveConnectionProvider, true); } return this.reactiveConnection; }}StatefulConnection<E, E> getConnection() { synchronized (this.connectionMonitor) { // 第一次通过 getNativeConnection() 获取连接；之后直接返回该连接 if (this.connection == null) { this.connection = getNativeConnection(); } if (getValidateConnection()) { validateConnection(); } return this.connection; }}分析以上源码，关键就在于 shareNativeConnection 默认为 true，导致只有一个连接。更改 shareNativeConnection 的值为 true，并开启 lettuce 连接池，最大连接数设置为 6；再次测试，[root@eureka2 jstack]# ss | grep 6379tcp ESTAB 0 0 ::ffff:10.201.0.27:48937 ::ffff:10.201.0.30:6379tcp ESTAB 0 0 ::ffff:10.201.0.27:35842 ::ffff:10.201.0.30:6379tcp ESTAB 0 0 ::ffff:10.201.0.27:48932 ::ffff:10.201.0.30:6379tcp ESTAB 0 0 ::ffff:10.201.0.27:48930 ::ffff:10.201.0.30:6379tcp ESTAB 0 0 ::ffff:10.201.0.27:48936 ::ffff:10.201.0.30:6379tcp ESTAB 0 0 ::ffff:10.201.0.27:48934 ::ffff:10.201.0.30:6379[root@eureka2 jstack]# jstack 23080 | grep lettuce-epollEventLoop"lettuce-epollEventLoop-4-6" #69 daemon prio=5 os_prio=0 tid=0x00007fcfa4012000 nid=0x5af2 runnable [0x00007fcfa81ef000]“lettuce-epollEventLoop-4-5” #67 daemon prio=5 os_prio=0 tid=0x00007fcf94003800 nid=0x5af0 runnable [0x00007fcfa83f1000]“lettuce-epollEventLoop-4-4” #60 daemon prio=5 os_prio=0 tid=0x00007fcfa0003000 nid=0x5ae9 runnable [0x00007fcfa8af8000]“lettuce-epollEventLoop-4-3” #59 daemon prio=5 os_prio=0 tid=0x00007fcfb00b8000 nid=0x5ae8 runnable [0x00007fcfa8bf9000]“lettuce-epollEventLoop-4-2” #58 daemon prio=5 os_prio=0 tid=0x00007fcf6c00f000 nid=0x5ae7 runnable [0x00007fcfa8cfa000]“lettuce-epollEventLoop-4-1” #43 daemon prio=5 os_prio=0 tid=0x00007fcfac248800 nid=0x5a64 runnable [0x00007fd00c2b9000]可以看到已经建立了 6 个 redis 连接，并且创建了 6 个 eventLoop 线程。2.2 线程阻塞再次进行压力测试，结果如下：[root@hystrix-dashboard wrk]# wrk -t 10 -c 500 -d 30s –latency -T 3s -s post-test.lua ‘http://10.201.0.27:8888/api/v1/json’Running 30s test @ http://10.201.0.27:8888/api/v1/json 10 threads and 500 connections Thread Stats Avg Stdev Max +/- Stdev Latency 215.83ms 104.38ms 1.00s 75.76% Req/Sec 234.56 49.87 434.00 71.45% Latency Distribution 50% 210.63ms 75% 281.30ms 90% 336.78ms 99% 519.51ms 69527 requests in 30.04s, 22.43MB readRequests/sec: 2314.14Transfer/sec: 764.53KB[root@eureka2 jstack]# top -Hp 23080top - 10:08:10 up 162 days, 12:31, 2 users, load average: 2.92, 1.19, 0.53Threads: 563 total, 9 running, 554 sleeping, 0 stopped, 0 zombie%Cpu(s): 50.5 us, 10.2 sy, 0.0 ni, 36.2 id, 0.1 wa, 0.0 hi, 2.9 si, 0.0 stKiB Mem : 7677696 total, 215924 free, 3308248 used, 4153524 buff/cacheKiB Swap: 6291452 total, 6291452 free, 0 used. 3468352 avail Mem PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND23280 root 20 0 7418804 1.3g 7404 R 42.7 17.8 0:54.75 java23272 root 20 0 7418804 1.3g 7404 S 31.1 17.8 0:44.63 java23273 root 20 0 7418804 1.3g 7404 S 31.1 17.8 0:44.45 java23271 root 20 0 7418804 1.3g 7404 R 30.8 17.8 0:44.63 java23282 root 20 0 7418804 1.3g 7404 S 30.5 17.8 0:44.96 java23119 root 20 0 7418804 1.3g 7404 R 24.8 17.8 1:27.30 java23133 root 20 0 7418804 1.3g 7404 R 23.8 17.8 1:29.55 java23123 root 20 0 7418804 1.3g 7404 S 23.5 17.8 1:28.98 java23138 root 20 0 7418804 1.3g 7404 S 23.5 17.8 1:44.19 java23124 root 20 0 7418804 1.3g 7404 R 22.8 17.8 1:32.21 java23139 root 20 0 7418804 1.3g 7404 R 22.5 17.8 1:29.49 java最终结果没有任何提升，cpu 利用率依然不超过 400%，tps 也还是在 2300 request/s；单个 cpu 利用率最高不超过 50%，说明这次的瓶颈不是 cpu。通过 jstack 查看线程状态，“lettuce-epollEventLoop-4-3” #59 daemon prio=5 os_prio=0 tid=0x00007fcfb00b8000 nid=0x5ae8 waiting for monitor entry [0x00007fcfa8bf8000] java.lang.Thread.State: BLOCKED (on object monitor) at org.springframework.data.redis.core.script.DefaultRedisScript.getSha1(DefaultRedisScript.java:88) - waiting to lock <0x000000008c1da690> (a java.lang.Object) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor.eval(DefaultReactiveScriptExecutor.java:113) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor.lambda$execute$0(DefaultReactiveScriptExecutor.java:105) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor$$Lambda$1317/1912229933.doInRedis(Unknown Source) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor.lambda$execute$6(DefaultReactiveScriptExecutor.java:167) at org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor$$Lambda$1318/1719274268.get(Unknown Source) at reactor.core.publisher.FluxDefer.subscribe(FluxDefer.java:46) at reactor.core.publisher.FluxDoFinally.subscribe(FluxDoFinally.java:73) at reactor.core.publisher.FluxOnErrorResume.subscribe(FluxOnErrorResume.java:47) at reactor.core.publisher.MonoReduceSeed.subscribe(MonoReduceSeed.java:65) at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59) at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60) at reactor.core.publisher.Mono.subscribe(Mono.java:3608) at reactor.core.publisher.FluxFlatMap.trySubscribeScalarMap(FluxFlatMap.java:169) at reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:53) at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1476) at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1476) at reactor.core.publisher.MonoProcessor.subscribe(MonoProcessor.java:457) at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1476) at reactor.core.publisher.MonoHasElement$HasElementSubscriber.onNext(MonoHasElement.java:74) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1476) at reactor.core.publisher.MonoProcessor.onNext(MonoProcessor.java:389) at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76) at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.onNext(FluxDoFinally.java:123) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) at reactor.core.publisher.FluxFilter$FilterSubscriber.onNext(FluxFilter.java:107) at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76) at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73) at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:238) at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onNext(FluxDefaultIfEmpty.java:92) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76) at io.lettuce.core.RedisPublisher$RedisSubscription.onNext(RedisPublisher.java:270) at io.lettuce.core.RedisPublisher$SubscriptionCommand.complete(RedisPublisher.java:754) at io.lettuce.core.protocol.CommandWrapper.complete(CommandWrapper.java:59) at io.lettuce.core.protocol.CommandHandler.complete(CommandHandler.java:646) at io.lettuce.core.protocol.CommandHandler.decode(CommandHandler.java:604) at io.lettuce.core.protocol.CommandHandler.channelRead(CommandHandler.java:556) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1434) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:965) at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799) at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:433) at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:330) at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:897) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.lang.Thread.run(Thread.java:748)有 4 个 lettuce-epollEventLoop 线程都处于 BLOCKED 状态，继续查看源码：public class DefaultRedisScript<T> implements RedisScript<T>, InitializingBean { private @Nullable ScriptSource scriptSource; private @Nullable String sha1; private @Nullable Class<T> resultType; public String getSha1() { // 1. 线程需要先获取 shaModifiedMonitor 锁 synchronized (shaModifiedMonitor) { // 第一次调用时或者 lua 脚本文件被修改时，需要重新计算 sha1 的值 // 否则直接返回sha1 if (sha1 == null || scriptSource.isModified()) { this.sha1 = DigestUtils.sha1DigestAsHex(getScriptAsString()); } return sha1; } } public String getScriptAsString() { try { return scriptSource.getScriptAsString(); } catch (IOException e) { throw new ScriptingException(“Error reading script text”, e); } }}public class ResourceScriptSource implements ScriptSource { // 只有第一次调用或者 lua 脚本文件被修改时，才会执行这个方法 @Override public String getScriptAsString() throws IOException { synchronized (this.lastModifiedMonitor) { this.lastModified = retrieveLastModifiedTime(); } Reader reader = this.resource.getReader(); return FileCopyUtils.copyToString(reader); } @Override public boolean isModified() { // 2. 每次都需要判断 lua 脚本是否被修改 // 线程需要再获取 lastModifiedMonitor 锁 synchronized (this.lastModifiedMonitor) { return (this.lastModified < 0 || retrieveLastModifiedTime() > this.lastModified); } }}对于限流操作，重要性并没有那么高，而且计算接口调用次数的 lua 脚本，一般也不会经常改动，所以没必要获取 sha1 的值的时候都查看下脚本是否有改动；如果偶尔改动的话，可以通过新增一个刷新接口，在改动脚本文件后通过手动刷新接口来改变 sha1 的值。所以这里，可以把同步操作去掉；我改成了这样：public class CustomRedisScript<T> extends DefaultRedisScript<T> { private @Nullable String sha1; CustomRedisScript(ScriptSource scriptSource, Class<T> resultType) { setScriptSource(scriptSource); setResultType(resultType); this.sha1 = DigestUtils.sha1DigestAsHex(getScriptAsString()); } @Override public String getSha1() { return sha1; }}2.3 cpu 出现大量软中断继续测试，结果如下：[root@hystrix-dashboard wrk]# wrk -t 10 -c 500 -d 30s -T 3s -s post-test.lua –latency ‘http://10.201.0.27:8888/api/v1/json’Running 30s test @ http://10.201.0.27:8888/api/v1/json 10 threads and 500 connections Thread Stats Avg Stdev Max +/- Stdev Latency 155.60ms 110.40ms 1.07s 67.68% Req/Sec 342.90 64.88 570.00 70.35% Latency Distribution 50% 139.14ms 75% 211.03ms 90% 299.74ms 99% 507.03ms 102462 requests in 30.02s, 33.15MB readRequests/sec: 3413.13Transfer/sec: 1.10MBcpu 利用率 500% 左右，tps 达到了 3400 req/s，性能大幅度提升。查看 cpu 状态：[root@eureka2 imf2]# top -Hp 19021top - 16:24:09 up 163 days, 18:47, 2 users, load average: 3.03, 1.08, 0.47Threads: 857 total, 7 running, 850 sleeping, 0 stopped, 0 zombie%Cpu0 : 60.2 us, 10.0 sy, 0.0 ni, 4.3 id, 0.0 wa, 0.0 hi, 25.4 si, 0.0 st%Cpu1 : 64.6 us, 16.3 sy, 0.0 ni, 19.0 id, 0.0 wa, 0.0 hi, 0.0 si, 0.0 st%Cpu2 : 65.7 us, 15.8 sy, 0.0 ni, 18.5 id, 0.0 wa, 0.0 hi, 0.0 si, 0.0 st%Cpu3 : 54.5 us, 15.8 sy, 0.0 ni, 29.5 id, 0.3 wa, 0.0 hi, 0.0 si, 0.0 st%Cpu4 : 55.0 us, 17.8 sy, 0.0 ni, 27.2 id, 0.0 wa, 0.0 hi, 0.0 si, 0.0 st%Cpu5 : 53.2 us, 16.4 sy, 0.0 ni, 30.0 id, 0.3 wa, 0.0 hi, 0.0 si, 0.0 stKiB Mem : 7677696 total, 174164 free, 3061892 used, 4441640 buff/cacheKiB Swap: 6291452 total, 6291452 free, 0 used. 3687692 avail Mem PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND19075 root 20 0 7722156 1.2g 14488 S 41.4 15.9 0:55.71 java19363 root 20 0 7722156 1.2g 14488 R 40.1 15.9 0:41.33 java19071 root 20 0 7722156 1.2g 14488 R 37.1 15.9 0:56.38 java19060 root 20 0 7722156 1.2g 14488 S 35.4 15.9 0:52.74 java19073 root 20 0 7722156 1.2g 14488 R 35.1 15.9 0:55.83 javacpu0 利用率达到了 95.7%，几乎跑满。但是其中出现了 si（软中断）: 25.4%。查看软中断类型：[root@eureka2 imf2]# watch -d -n 1 ‘cat /proc/softirqs’ CPU0 CPU1 CPU2 CPU3 CPU4 CPU5 HI: 0 0 0 0 0 0 TIMER: 1629142082 990710808 852299786 606344269 586896512 566624764 NET_TX: 291570 833710 9616 5295 5358 2012064 NET_RX: 2563401537 32502894 31370533 6886869 6530120 6490002 BLOCK: 18130 1681 41404591 8751054 8695636 8763338BLOCK_IOPOLL: 0 0 0 0 0 0 TASKLET: 39225643 0 0 817 17304 2516988 SCHED: 782335782 442142733 378856479 248794679 238417109 259695794 HRTIMER: 0 0 0 0 0 0 RCU: 690827224 504025610 464412234 246695846 254062933 248859132其中 NET_RX，CPU0 的中断次数远远大于其他 CPU，初步判断是网卡问题。我这边网卡是 ens32，查看网卡的中断号：[root@eureka2 imf2]# cat /proc/interrupts | grep ens 18: 2524017495 0 0 0 0 7 IO-APIC-fasteoi ens32[root@eureka2 imf2]# cat /proc/irq/18/smp_affinity01[root@eureka2 imf2]# cat /proc/irq/18/smp_affinity_list0网卡的中断配置到了 CPU0。（01：表示 cpu0，02：cpu1，04：cpu2，08：cpu3，10：cpu4，20：cpu5）smp_affinity：16 进制；smp_affinity_list：配置到了哪些 cpu。查看网卡队列模式：[root@eureka2 ~]# lspci -vvv02:00.0 Ethernet controller: Intel Corporation 82545EM Gigabit Ethernet Controller (Copper) (rev 01) Subsystem: VMware PRO/1000 MT Single Port Adapter Physical Slot: 32 Control: I/O+ Mem+ BusMaster+ SpecCycle- MemWINV+ VGASnoop- ParErr- Stepping- SERR+ FastB2B- DisINTx- Status: Cap+ 66MHz+ UDF- FastB2B- ParErr- DEVSEL=medium >TAbort- <TAbort- <MAbort- >SERR- <PERR- INTx- Latency: 0 (63750ns min), Cache Line Size: 64 bytes Interrupt: pin A routed to IRQ 18 Region 0: Memory at fd5c0000 (64-bit, non-prefetchable) [size=128K] Region 2: Memory at fdff0000 (64-bit, non-prefetchable) [size=64K] Region 4: I/O ports at 2000 [size=64] [virtual] Expansion ROM at fd500000 [disabled] [size=64K] Capabilities: [dc] Power Management version 2 Flags: PMEClk- DSI+ D1- D2- AuxCurrent=0mA PME(D0+,D1-,D2-,D3hot+,D3cold+) Status: D0 NoSoftRst- PME-Enable- DSel=0 DScale=1 PME- Capabilities: [e4] PCI-X non-bridge device Command: DPERE- ERO+ RBC=512 OST=1 Status: Dev=ff:1f.0 64bit+ 133MHz+ SCD- USC- DC=simple DMMRBC=2048 DMOST=1 DMCRS=16 RSCEM- 266MHz- 533MHz- Kernel driver in use: e1000 Kernel modules: e1000由于是单队列模式，所以通过修改 /proc/irq/18/smp_affinity 的值不能生效。可以通过 RPS/RFS 在软件层面模拟多队列网卡功能。[root@eureka2 ~]# echo 3e > /sys/class/net/ens32/queues/rx-0/rps_cpus[root@eureka2 rx-0]# sysctl net.core.rps_sock_flow_entries=32768[root@eureka2 rx-0]# echo 32768 > /sys/class/net/ens32/queues/rx-0/rps_flow_cnt/sys/class/net/ens32/queues/rx-0/rps_cpus: 1e，设置模拟网卡中断分配到 cpu1-5 上。继续测试，[root@hystrix-dashboard wrk]# wrk -t 10 -c 500 -d 30s -T 3s -s post-test.lua –latency ‘http://10.201.0.27:8888/api/v1/json’Running 30s test @ http://10.201.0.27:8888/api/v1/json 10 threads and 500 connections Thread Stats Avg Stdev Max +/- Stdev Latency 146.75ms 108.45ms 1.01s 65.53% Req/Sec 367.80 64.55 575.00 67.93% Latency Distribution 50% 130.93ms 75% 200.72ms 90% 290.32ms 99% 493.84ms 109922 requests in 30.02s, 35.56MB readRequests/sec: 3661.21Transfer/sec: 1.18MB[root@eureka2 rx-0]# top -Hp 19021top - 09:39:49 up 164 days, 12:03, 1 user, load average: 2.76, 2.02, 1.22Threads: 559 total, 9 running, 550 sleeping, 0 stopped, 0 zombie%Cpu0 : 55.1 us, 13.0 sy, 0.0 ni, 17.5 id, 0.0 wa, 0.0 hi, 14.4 si, 0.0 st%Cpu1 : 60.1 us, 14.0 sy, 0.0 ni, 22.5 id, 0.0 wa, 0.0 hi, 3.4 si, 0.0 st%Cpu2 : 59.5 us, 14.3 sy, 0.0 ni, 22.4 id, 0.0 wa, 0.0 hi, 3.7 si, 0.0 st%Cpu3 : 58.6 us, 15.2 sy, 0.0 ni, 22.2 id, 0.0 wa, 0.0 hi, 4.0 si, 0.0 st%Cpu4 : 59.1 us, 14.8 sy, 0.0 ni, 22.7 id, 0.0 wa, 0.0 hi, 3.4 si, 0.0 st%Cpu5 : 57.7 us, 16.2 sy, 0.0 ni, 23.0 id, 0.0 wa, 0.0 hi, 3.1 si, 0.0 stKiB Mem : 7677696 total, 373940 free, 3217180 used, 4086576 buff/cacheKiB Swap: 6291452 total, 6291452 free, 0 used. 3533812 avail Mem PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND19060 root 20 0 7415812 1.2g 13384 S 40.7 16.7 3:23.05 java19073 root 20 0 7415812 1.2g 13384 R 40.1 16.7 3:20.56 java19365 root 20 0 7415812 1.2g 13384 R 40.1 16.7 2:36.65 java可以看到软中断也分配到了 cpu1-5 上；至于为什么还是 cpu0 上软中断比例最高，猜测是因为还有一些其他中断并且默认配置在 cpu0 上？同时，tps 也从 3400 -> 3600，提升不大。2.4 增加 redis 连接经过以上修改，cup 利用率还是不超过 500%，说明在某些地方还是存在瓶颈。尝试修改了下 lettuce 连接池，spring: redis: database: x host: x.x.x.x port: 6379 lettuce: pool: max-active: 18 min-idle: 1 max-idle: 18主要是把 max-active 参数 6 增大到了 18，继续测试：[root@hystrix-dashboard wrk]# wrk -t 10 -c 500 -d 120s -T 3s -s post-test.lua –latency ‘http://10.201.0.27:8888/api/v1/json’Running 2m test @ http://10.201.0.27:8888/api/v1/json 10 threads and 500 connections Thread Stats Avg Stdev Max +/- Stdev Latency 117.66ms 96.72ms 1.34s 86.48% Req/Sec 485.42 90.41 790.00 70.80% Latency Distribution 50% 90.04ms 75% 156.01ms 90% 243.63ms 99% 464.04ms 578298 requests in 2.00m, 187.01MB readRequests/sec: 4815.57Transfer/sec: 1.56MB6 核 cpu 几乎跑满，同时 tps 也从 3600 -> 4800，提升明显！这说明之前的瓶颈出在 redis 连接上，那么如何判断 tcp 连接是瓶颈呢？（尝试通过 ss、netstat 等命令查看 tcp 发送缓冲区、接收缓冲区、半连接队列、全连接队列等，未发现问题。先放着，以后在研究）

在我们讨论系统稳定性的时候，其实核心的关键词就是容量规划，如何做好业务容量与系统性能的评估，是保障系统稳定性的关键。对于系统性能的评估，需要我们具备自动化工具来对系统进行性能压测，探测系统在实际业务场景下的基线数据，这是我们进行系统资源配置的基础，也是在应对流量增长时进行弹性扩容的依据。在我们做好容量规划的前提下，在实际业务场景下，我们还是不可避免的会面对不确定的系统压力，在面对突发不确定流量的情况下，我们最担心的就是系统的“雪崩”。就像突然爆发的车流让道路交通瘫痪一样，我们的系统在突发流量下，很可能像多米诺骨牌一样，全链路的崩塌。很多情况下，我们以为我们的系统能够这样：但实际上确实这样在系统发生雪崩的情况下，我们连基本容量规划的负载都保证不了。如何保障我们的系统能够在复杂多变的业务场景下，能够持续稳定的提供的负载处理能力。这就要求我们按照系统的容量规划，做好系统的限流保护，让超出负载的流量能够快速的failover，将系统无法承载的流量拒之门外，保护我们的系统不会发生雪崩。如何做限流在一般的概念中，我们在讨论限流的时候，首先会想到并发的限流，通过限制系统服务调用的并发数来对系统进行保护。这里面其实包含两层概念：线程数和QPS。线程数是我们系统中实际活跃的处理线程的数目，而QPS是我们系统在一定时间度量范围内的访问速率。一般传统的限流手段是通过对活跃线程数进行控制来进行系统流量的控制。在现今复杂的互联网服务调用体系下，更常用的是QPS进行限制来达到我们系统保护的效果。考虑我们现今大部分的互联网架构，一般业务系统会由几十甚至上百的微服务构成，一次业务请求的处理，涉及众多的微服务调用。从线程数的角度看，很难去对整体系统的流量做一个衡量。在我们对复杂的业务系统进行性能评估的时候，会使用全链路压测的工具来进行，衡量的目标也是基于系统的QPS.所以使用对QPS进行限制的手段，会更好的对业务系统进行一个负载保护。在笔者多年稳定性的负责工作中，一般在系统的各个入口处使用QPS作为主要的限流保护手段，线程数限流更多是作为上游系统对下游系统的一种熔断机制来使用，保护上游系统在下游RT变长的情况下，不会被拖垮。限流的实现线程数限流对于线程数的限流，一般通过令牌许可的方式实现，通过预置令牌的数目来控制系统的活跃线程数量，没有获取令牌的线程，快速的失败返回。在线程的入口处获取许可令牌，在执行业务逻辑完毕的出口处归还令牌。在实现上一般使用try{……}finally{……}结构实现，在finally中进行令牌的释放。线程数的限流多数用于简单系统的负载保护，及自我的熔断保护优雅降级，防止依赖的下游服务RT变长的情况下，造成自身系统性能下降。在例如天猫交易系统这种，依赖下游服务（库存、物流、商品、营销等）较多的系统中，一个服务的变慢可能会对处于核心的交易系统造成不良的影响，所以对于在交易系统中，对于依赖的核心服务都是设置一个自我保护的线程数限流值，在下游服务出问题的情况下进行优雅降级。QPS限流和线程数不同，QPS限流是对按照一定时间单位内的并发速率来对系统进行限制。QPS与线程数限流相比，最大的好处除了能够按照压测模型的测算进行设置外，在笔者看来，另一个最大的好处就是可以对流量进行削峰，防止流量突刺对系统资源进行穿透影响。QPS虽然度量单位上是秒，但是一般的实现不会基于秒级进行计数统计来进行限流保护，基于秒级的计数时间窗口过大，不能够消除流量突刺。试想一下在一秒的时间窗口内，流量突发集中在后面10ms，那么在最后的时刻，系统流量会变成我们期望值的100倍，可想而知这个时候对于系统来说是多么可怕的灾难。所以在实现上，一般基于时间窗口的分片来进行计数，将整个QPS的计数周期划分为多个时间窗口，将计数值分散在各个时间窗口进行控制和计算，这样能够避免上面提到的流量突刺的出现，能够有效的将流量进行削峰，将流量洪峰削平，放着洪峰对于系统资源的破坏，例如缓存的热点穿透在软件的世界里很奇妙，我们会发现很多东西没有银弹，软件的进化过程，就是跟着实际的场景不断的调优。基于时间窗口的QPS限流，虽然能够消除流量突刺的场景，但是不可避免的会造成流量的损耗。试想一下如果流量的分布不均匀，那么在流量少的时间窗口下，没有到达的流量实际是损耗的。在实际场景中，例如天猫双十一零点高峰，流量实际在每个时间窗口都是处于极度饱和的状态，这个时候简单的窗口分片就能达到很好的限流效果。在流量分布不均匀的情况下，我们可以实现动态的时间窗口分片，将不饱和的时间窗口流量在不超过系统负荷的情况下累加到下一时间窗口，这样可以降低流量的损耗。Pontus-用友云限流服务用友云iuap平台为了保证客户云端系统服务的稳定性，结合微服务平台的建设，推出了稳定服务套件之一的限流服务Pontus，帮助用友云客户构建稳定性的系统防护体系。Pontus支持QPS和线程限流两种方式，结合用友云微服务治理平台，将限流服务通过中间件统一服务的方式，内置在应用的微服务框架体系中，对微服务接口提供统一的限流一级防护。采用动态滑动窗口机制，能够对流量峰值进行削峰，并且最大程度的降低流量损耗。触发限流及后端接口响应超时时，自动触发接口降级。pontus服务与用友云开发者中心无缝集成，包含在用友云中间件统一SDK中，用户应用在开发者中心部署之后，能够自动发现服务接口，并对相应服务接口进行限流配置。源于电商大促多年实践场景，Pontus能够为云应用提供稳定一致的流量防护，进行流量削峰和系统熔断，为云应用的稳定性保驾护航。