一文搞清楚限流、熔断和降级

APP运营 2023-4-23 17:39 8490人围观限流

限流

是对系统的被请求频次以及内部的部分功用的履行频次加以限制，避免因突发的流量激增，致使全部系统不成用。限流主如果防御庇护手段，从流量泉源起头控制流量躲避题目。

熔断

是在流量过大时（或下流办事出现题目时），可以自动断开与下流办事的交互，并可以经过自我诊断下流系统的毛病能否已经批改，或上游流量能否削减至一般水平，来规复自我规复。熔断更像是自动化解救手段，能够发生在办事没法支持大量请求或办事发生其他故障时，对请求停止限制处置，同时还可尝试性的停止规复。

办事升级

主如果针对非焦点营业功用，而焦点营业假如流程跨越预估的峰值，就需要停止限流。升级一般斟酌的是散布式系统的整体性，从泉源上切断流量的来历。升级更像是预估手段，在估计流量峰值条件下，提早经过设置功用下降办事体验，或停息主要功用，保证系统首要流程功用平稳响应。

限流和熔断也可以看做是一种办事升级的手段。以下对每个概念停止具体味商。

什么是限流？

限流是流量限速（Rate Limit）的简称，是指只答应指定的事务进入系统，跨越的部分将被拒绝办事、排队或期待、升级等处置。对于server办事而言，限流为了保证一部分的请求流量可以获得一般的响应，总好过全数的请求都不能获得响应，甚至致使系统雪崩。

限流常见的算法：

1.计数器算法（牢固窗口）

在一段时候间隔内（时候窗）处置请求的最大数目牢固，跨越部分不做处置。

计数器算法Java简单实现

// 计速器 限速@Slf4jpublic class CounterLimiter {    // 肇端时候    private static long lastTime = System.currentTimeMillis();    // 时候区间的时候间隔 ms    private static long interval = 1000;    // 每秒限制数目    private static long capacity = 2;    //累加器    private static AtomicLong counter = new AtomicLong();    //返回正数，暗示没有被限流    //返回-1 ，暗示被限流    // 计数判定, 能否超越限制    private static long tryPass(long requestId, int turn) {        long nowTime = System.currentTimeMillis();        //当前时候，在时候区间之内        if (nowTime < lastTime + interval) {            //增加计数            long count = counter.incrementAndGet();            if (count <= capacity) {                return count;            } else {                return -1;  // 容量耗尽            }        } else {            //在时候区间之外            synchronized (CounterLimiter.class) {                log.info("新时候区到了,requestId{}, turn {}..", requestId, turn);                // 再一次判定，避免反复初始化                // dc 两重校验                if (nowTime > lastTime + interval) {                    //reset counter                    counter.set(1);                    lastTime = nowTime;                }else {                    //补充上遗漏的场景， 计数器算法看似简单，现实上圈套重重                    //增加计数                    long count = counter.incrementAndGet();                    if (count <= capacity) {                        return count;                    } else {                        return -1;  // 容量耗尽                    }                }            }            return 1;        }    }    //线程池，用于多线程模拟测试    private ExecutorService pool = Executors.newFixedThreadPool(10);    @Test    public void testLimit() {        // 被限制的次数        AtomicInteger limited = new AtomicInteger(0);        // 线程数        final int threads = 2;        // 每条线程的履行轮数        final int turns = 20;        // 同步器        CountDownLatch countDownLatch = new CountDownLatch(threads);        long start = System.currentTimeMillis();        for (int i = 0; i < threads; i++) {            pool.submit(() ->            {                try {                    for (int j = 0; j < turns; j++) {                        long taskId = Thread.currentThread().getId();                        long index = tryPass(taskId, j);                        System.out.println("index = " + index);                        if (index <= 0) {                            // 被限制的次数积累                            limited.getAndIncrement();                        }                        Thread.sleep(200);                    }                } catch (Exception e) {                    e.printStackTrace();                }                //期待一切线程竣事                countDownLatch.countDown();            });        }        try {            countDownLatch.await();        } catch (InterruptedException e) {            e.printStackTrace();        }        float time = (System.currentTimeMillis() - start) / 1000F;        //输出统计成果        log.info("限制的次数为：" + limited.get() +                ",经过的次数为：" + (threads * turns - limited.get()));        log.info("限制的比例为：" + (float) limited.get() / (float) (threads * turns));        log.info("运转的时长为：" + time);    }}

计数器算法的弱点

在两个间隔之间，倘使有麋集的请求。则会致使单元时候内的现实请求跨越阈值。

例如，限制每分钟100个请求，在第一个1分钟时候窗里临界点的时辰（比如 0:59时辰），瞬间来了 100 个请求，这时可以一般处置请求，然后在1:01时辰的时辰，第二个1分钟时候窗里又来了100个请求，这时也可以一般处置这些请求。但在 3s 内，一共处置 200 个请求，可是时候跨度小于1分钟，能够会形成后端过载。

一文搞清楚限流、熔断和升级

所以在利用计数器限流战略时，用户经过在时候窗口的毗连处突发大量请求，可以瞬间跨越速度限制，从而瞬间压垮利用，这就是计数器限流算法的临界题目。

若何处置临界题目，可以经过滑动窗口、漏铜、令牌桶三种算法处理。

2.滑动窗口算法

根基概念：

滑动窗口和时候桶

滑动窗口 可以这么来了解滑动窗口：一位乘客坐在正在行驶的列车的靠窗座位上，列车行驶的公路两侧种着一排挺拔的白杨树，随着列车的进步，路边的白杨树敏捷从窗口滑过，我们用每棵树来代表一个请求，用列车的行驶代表时候的流逝，那末，列车上的这个窗口就是一个典型的滑动窗口，这个乘客能经过窗口看到的白杨树的数目，就是滑动窗口要统计的数据。
时候桶 时候桶是统计滑动窗口数据时的最小单元。一样类比列车窗口，在列车速度很是快时，假如每擦过一棵树就统计一次窗口内树的数据，明显开销很是大，假如乘客将窗口分红N份，进步行时列车每擦过窗口的N分之一就统计一次数据，开销就大大地减小了。简单来说，时候桶也就是滑动窗口的N分之一。

滑动窗口算法填补了计数器算法的不敷。

滑动窗口算法把间隔时候分别红更小的粒度，当更小粒度的时候间隔曩昔后，把曩昔的间隔请求数减掉，再补充一个空的时候间隔。

以下图所示，把1分钟分别为10个更小的时候间隔，每6s为一个间隔。每个间隔是1个时候桶

一文搞清楚限流、熔断和升级

一个时候窗口为1分钟，滑动窗口分红10个时候桶，每个时候桶6秒。
每过6秒，滑动窗口向右移动1个时候桶。
每个时候桶都有自力的计数器。
假如时候窗口内一切时候桶的计数器之和跨越了限流阀值，则触发限流操纵。

以下图所示，滑动窗口算法比计数器算法控制得更邃密。

一文搞清楚限流、熔断和升级

滑动窗口算法伪代码

/**     * 单元时候分别的小周期（单元时候是1分钟，10s一个小格子窗口，一共6个格子）     */    private int SUB_CYCLE = 10;    /**     * 每分钟限流请求数     */    private int thresholdPerMin = 100;    /**     * 计数器, k-为当前窗口的起头时候值秒，value为当前窗口的计数     */    private final TreeMap<Long, Integer> counters = new TreeMap<>();   /**     * 滑动窗口时候算法实现     */    boolean slidingWindowsTryAcquire() {        long currentWindowTime = LocalDateTime.now().toEpochSecond(ZoneOffset.UTC) / SUB_CYCLE * SUB_CYCLE; //获得当前时候在哪个小周期窗口        int currentWindowNum = countCurrentWindow(currentWindowTime); //当前窗口总请求数        //跨越阀值限流        if (currentWindowNum >= thresholdPerMin) {            return false;        }        //计数器+1        counters.get(currentWindowTime)++;        return true;    }   /**    * 统计当前窗口的请求数    */    private int countCurrentWindow(long currentWindowTime) {        //计较窗口起头位置        long startTime = currentWindowTime - SUB_CYCLE* (60s/SUB_CYCLE-1);        int count = 0;        //遍历存储的计数器        Iterator<Map.Entry<Long, Integer>> iterator = counters.entrySet().iterator();        while (iterator.hasNext()) {            Map.Entry<Long, Integer> entry = iterator.next();            // 删除无效过期的子窗口计数器            if (entry.getKey() < startTime) {                iterator.remove();            } else {                //累加当前窗口的一切计数器之和                count =count + entry.getValue();            }        }        return count;    }

弱点

滑动窗口设备得越邃密，限流的结果越好，但滑动窗口的时候间隔（小格子）多了，存储的空间也会增加。

3.漏铜桶算法

请求间接进入到漏桶里，漏桶以一定的速度对请求停止放行，当请求数目递增，漏桶内的总请求量大于桶容量就会间接溢出，请求被拒绝。

一文搞清楚限流、熔断和升级

漏桶限流法则 1）请求以肆意速度流入漏桶。 2）漏桶的容量是牢固的，放行速度也是牢固的。 3）漏桶容量是稳定的，假如处置速度太慢，桶内请求数目会超越桶的容量，则前面流入的请求会溢出，暗示请求被拒绝。

public class LeakBucketLimiter {    // 计较的肇端时候    private static long lastOutTime = System.currentTimeMillis();    // 时候区间的时候间隔 ms    private static long interval = 1000;    // 流出速度 每秒 2 次    private static int leakRate = 2;    // 桶的容量    private static int capacity = 20;    //残剩的容量    private static AtomicInteger waterInBucket = new AtomicInteger(0);    //返回值说明：    // false 没有被限制到    // true 被限流    public static synchronized boolean isLimit(long taskId, int turn) {        // 假如是空桶，就当前时候作为漏出的时候        if (waterInBucket.get() == 0) {            lastOutTime = System.currentTimeMillis();            waterInBucket.addAndGet(1);            return false;        }        //场景三：当前请求和上次请求，在同一个时候区间        long nowTime = System.currentTimeMillis();        //当前时候，在时候区间之内        //漏水以时候区间为计较维度，同一个区间，没有需要反复去计较漏水        if (nowTime < lastOutTime + interval) {            // 尝试加水,而且水还未满 ，放行            if ((waterInBucket.get()) < capacity) {                waterInBucket.addAndGet(1);                return false;            } else {                // 水满，拒绝加水， 限流                return true;            }        }        //场景二： 桶里边有水        //当前时候，在时候区间之外        // 计较漏水，以时候的区间为维度的        int waterLeaked = ((int) ((System.currentTimeMillis() - lastOutTime) / 1000)) * leakRate;        // 计较残剩水量        int waterLeft = waterInBucket.get() - waterLeaked;        //校正数据        waterLeft = Math.max(0, waterLeft);        waterInBucket.set(waterLeft);        // 重新更新leakTimeStamp        lastOutTime = System.currentTimeMillis();        // 尝试增加请求,而且容量还未满 ，放行        if ((waterInBucket.get()) < capacity) {            waterInBucket.addAndGet(1);            return false;        } else {            // 漏桶满了，限流            return true;        }    }    //线程池，用于多线程模拟测试    private ExecutorService pool = Executors.newFixedThreadPool(10);    @Test    public void testLimit() {        // 被限制的次数        AtomicInteger limited = new AtomicInteger(0);        // 线程数        final int threads = 2;        // 每条线程的履行轮数        final int turns = 20;        // 线程同步器        CountDownLatch countDownLatch = new CountDownLatch(threads);        long start = System.currentTimeMillis();        for (int i = 0; i < threads; i++) {            pool.submit(() ->            {                try {                    for (int j = 0; j < turns; j++) {                        long taskId = Thread.currentThread().getId();                        boolean intercepted = isLimit(taskId, j);                        if (intercepted) {                            // 被限制的次数积累                            limited.getAndIncrement();                        }                        Thread.sleep(200);                    }                } catch (Exception e) {                    e.printStackTrace();                }                //期待一切线程竣事                countDownLatch.countDown();            });        }        try {            countDownLatch.await();        } catch (InterruptedException e) {            e.printStackTrace();        }        float time = (System.currentTimeMillis() - start) / 1000F;        //输出统计成果        log.info("限制的次数为：" + limited.get() +                ",经过的次数为：" + (threads * turns - limited.get()));        log.info("限制的比例为：" + (float) limited.get() / (float) (threads * turns));        log.info("运转的时长为：" + time);    }}

弱点

漏桶的出水速度是牢固的，也就是请求放行速度是牢固的，故漏桶不能有用应对突发流量，可是能起到平滑突发流量（整流）的感化。

4.令牌桶算法

令牌桶算法以一个设定的速度发生令牌并放入令牌桶，每次用户请求都得申请令牌，假如令牌不敷，则拒绝请求。

一文搞清楚限流、熔断和升级

令牌的数目与时候和发放速度强相关，流逝的时候越长，往桶里加入令牌的越多，假如令牌发放速度比申请速度快，则令牌会放入令牌桶，直到占满全部令牌桶，令牌桶满了，多的令牌就间接抛弃。

令牌桶限流大致的法则以下

1）进水口依照某个速度向桶中放入令牌。

2）令牌桶的容量是牢固的，可是放行的速度不是牢固的，只要桶中还有残剩令牌，一旦请求过来就能申请成功，然后放行。

3）假如令牌的发放速度慢于请求的到来速度，则桶内就无牌可领，请求就会被拒绝。

令牌桶算法的Java简单实现

public class TokenBucketLimiter {    // 上一次令牌发放时候    public long lastTime = 0;    // 桶的容量    public int capacity = 2;    // 令牌天生速度 /s    public int rate = 2;    // 当前令牌数目    public AtomicInteger tokens = new AtomicInteger(0);    //返回值说明：    // false 没有被限制到    // true 被限流    public synchronized boolean isLimited(long taskId, int APPlyCount) {        long now = System.currentTimeMillis();        //时候间隔,单元为 ms        long gap = now - lastTime;        //场景三：当前请求和上次请求，在同一个时候区间        //当前时候，在时候区间之内        //以时候区间为计较维度，同一个区间，没有需要反复去计较令牌数目        if (lastTime != 0 && gap < 1000/*interval*/) {            if (tokens.get() < applyCount) {                // 若拿不到令牌,则拒绝                // log.info("被限流了.." + taskId + ", applyCount: " + applyCount);                return true;            } else {                // 还有令牌，支付令牌                tokens.getAndAdd(-applyCount);                // log.info("残剩令牌.." + tokens);                return false;            }        }        System.out.println("时区之外 gap = " + gap);        if (lastTime == 0) {            gap = 1000 /*interval*/;        }        //计较时候段内的令牌数        int reverse_permits = (int) (gap * rate / 1000);        int all_permits = tokens.get() + reverse_permits;        // 当前令牌数        tokens.set(Math.min(capacity, all_permits));        log.info("tokens {} capacity {} gap {} ", tokens, capacity, gap);        lastTime = now;        if (tokens.get() < applyCount) {            // 若拿不到令牌,则拒绝            // log.info("被限流了.." + taskId + ", applyCount: " + applyCount);            return true;        } else {            // 还有令牌，支付令牌            tokens.getAndAdd(-applyCount);            // log.info("残剩令牌.." + tokens);            return false;        }    }    //线程池，用于多线程模拟测试    private ExecutorService pool = Executors.newFixedThreadPool(10);    @Test    public void testLimit() {        // 被限制的次数        AtomicInteger limited = new AtomicInteger(0);        // 线程数        final int threads = 2;        // 每条线程的履行轮数        final int turns = 20;        // 同步器        CountDownLatch countDownLatch = new CountDownLatch(threads);        long start = System.currentTimeMillis();        for (int i = 0; i < threads; i++) {            pool.submit(() ->            {                try {                    for (int j = 0; j < turns; j++) {                        long taskId = Thread.currentThread().getId();                        boolean intercepted = isLimited(taskId, 1);                        if (intercepted) {                            // 被限制的次数积累                            limited.getAndIncrement();                        }                        Thread.sleep(200);                    }                } catch (Exception e) {                    e.printStackTrace();                }                //期待一切线程竣事                countDownLatch.countDown();            });        }        try {            countDownLatch.await();        } catch (InterruptedException e) {            e.printStackTrace();        }        float time = (System.currentTimeMillis() - start) / 1000F;        //输出统计成果        log.info("限制的次数为：" + limited.get() +                ",经过的次数为：" + (threads * turns - limited.get()));        log.info("限制的比例为：" + (float) limited.get() / (float) (threads * turns));        log.info("运转的时长为：" + time);    }}

什么是熔断

1. 熔断

1.1 熔断来历

辞书诠释：保险丝烧断保险丝也被称为熔断器。

在90年月很多电闸都有保险丝，现在都是空气开关，当电压出现短路题目时，保险丝过热，由因而铅丝熔点低于铜铁，所以短路形成线路温度升高，致使铅丝融化，线路断开。现在电路自动断开，我们的电器就会遭到到庇护。否则，能够形成火灾等严重结果。

保险丝就是一个自我庇护装配，庇护全部电路。此概念被引入多个行业，我们方法会的是散布式系统中的熔断。

1.2 散布式系统中的熔断

在散布式系统中，分歧的办事相互依靠，某些办事需要依靠下流办事，下流办事非论是内部系统还是第三方办事，假如出现题目，我们还是自觉地延续地去请求，即使失利了屡次，还是不竭的去请求，去期待，便能够形成系统更严重的题目。

期待增加了全部链路的请求时候。
下流系统有题目，不竭的请求致使下流系统有延续的拜候压力难以规复。
期待时候太长形成上游办事线程阻塞，CPU被拉满等题目。
能够致使办事雪崩。

1.3 熔断的感化

熔断形式可以避免利用法式不竭地尝试请求下流能够超时和失利的办事，可以不必期待下流办事的修复而到达利用法式可履行的状态。

1.4为什么要利用熔断器

熔断器形式最重要的是能让利用法式自我诊断下流系统的毛病能否已经批改，假如没有，不放量去请求，假如请求成功了，渐渐的增加请求，再次尝试挪用。

熔断有三种状态：

1.Closed：封闭状态

一切请求都一般拜候。

2.Open：翻开状态 一切请求城市被升级

在封闭状态下，熔断器会对请讨情况计数，当一按时候内失利请求百分比到达阈值，则触发熔断，断路器会完全翻开。一般默许失利比例的阈值是50%，请求次数最少不低于20次。

3.Half Open：半开状态 答应部分请求经过

open状态不是永久的，在熔断器开启状态翻开后会进入休眠时候（一般默许是5S）。随后断路器会自动进入半开状态。此时会开释部分请求经过，若这些请求都是健康的，则会完全封闭断路器，否则继续连结翻开，再次停止休眠计时。

一文搞清楚限流、熔断和升级

熔断器开源组件

1.Hystrix

Hystrix（豪猪--->身上很多刺--->庇护自己），宣言“defend your app”，是由Netflflix开源的一个提早和容错库，用于隔离拜候远程系统、办事大概第三方库，避免级联失利，从而提升系统的可用性与容错性。Hystrix首要经过以下几点实现提早和容错：

包裹请求：利用HystrixCommand包裹对依靠的挪用逻辑。自动投递微办事方法(@HystrixCommand 增加Hystrix控制) ——挪用简历微办事

跳闸机制：当某办事的毛病率跨越一定的阈值时，Hystrix可以跳闸，停止请求该办事一段时候。

资本隔离：Hystrix为每个依靠都保护了一个小型的线程池(舱壁形式)(大概信号量)。假如该线程池已满，发往该依靠的请求就被立即拒绝，而不是排队等待，从而加速失利判定。

监控：Hystrix可以近乎实时地监控运转目标和设置的变化，例如成功、失利、超时、以及被拒绝的请求等。

回退机制：当请求失利、超时、被拒绝，或当断路器翻开时，履行回退逻辑。回退逻辑由开辟职员自行供给，例如返回一个缺省值。

自我修复：断路器翻开一段时候后，会自动进入“半开”状态。

2.Resilience4j

Resilience4j是一款轻量级，易于利用的容错库，其灵感来自于Netflix Hystrix，可是专为Java8和函数式编程而设想。轻量级，由于库只利用了Vavr，它没有任何其他内部依靠下。相比之下，Netflix Hystrix对Archaius具有编译依靠性，Archaius具有更多的内部库依靠性，例如Guava和Apache Commons Configuration。

Resilience4j与Hystrix首要区分：

1.Hystrix挪用必须被封装到HystrixCommand里，Resilience4j以装潢器的方式供给对函数式接口、lambda表达式等的嵌套装潢，是以你可以用简洁的方式组合多种高可用机制。

2.Hystrix采用滑动窗口方式统计频次，Resilience4j采用环形缓冲区

3.半开启状态下，Hystrix进利用一次履行判定能否停止状态转换，Resilience4j则可设置履行次数与阈值，经过设置参数履行判定能否停止状态转换，这类方式进步了熔断机制的稳定性。

4.Hystrix采用基于线程池和信号量的隔离，而resilience4j只供给基于信号量的隔离

3.Sentinel

随着微办事的风行，办事和办事之间的稳定性变得越来越重要。Sentinel 以流量为切入点，从流量控制、流量路由、熔断升级、系统自顺应过载庇护、热门流量防护等多个维度庇护办事的稳定性。

Sentinel 具有以下特征:

丰富的利用处景：Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的焦点场景，例如秒杀（即突发流量控制在系统容量可以承受的范围）、消息削峰填谷、集群流量控制、实时熔断下流不成用利用等。
完整的实时监控：Sentinel 同时供给实时的监控功用。您可以在控制台中看到接入利用的单台机械秒级数据，甚至 500 台以下范围的集群的汇总运转情况。
普遍的开源生态：Sentinel 供给开箱即用的与别的开源框架/库的整合模块，例如与 Spring Cloud、Apache Dubbo、gRPC、Quarkus 的整合。您只需要引入响应的依靠并停止简单的设置即可快速地接入 Sentinel。同时 Sentinel 供给 Java/Go/C++ 等多说话的原生实现。
完善的 SPI 扩大机制：Sentinel 供给简单易用、完善的 SPI 扩大接口。您可以经过实现扩大接口来快速地定制逻辑。例如定制法则治理、适配静态数据源等。

Hystrix健康统计滑动窗口的履行流程：

1）HystrixCommand号令器的履行成果（失利、成功）会以事务的形式经过RxJava事务流弹射进来，构成履行完成事务流。

2）桶计数流以事务流作为来历，将事务流中的事务依照固按时候长度（桶时候间隔）分别红转动窗口，并对时候桶转动窗口内的事务依照范例停止积累，完成以后将桶数据弹射进来，构成桶计数流。

3）桶滑动统计流以桶计数流作为来历，依照步长为1、长度为设定的桶数（设置的滑动窗口桶数）的法则分别滑动窗口，并对滑动窗口内的一切的桶数据依照各事务范例停止汇总，汇总成终极的窗口健康数据并弹射进来，构成终极的桶滑动统计流，作为Hystrix熔断器停止状态转换的数据支持。

模拟实现

public class WindowDemo{    /**     * window 建立操纵符 建立滑动窗口     * 演示 window 建立操纵符   建立滑动窗口     */    @Test    public void simpleWindowObserverDemo()    {        List<Integer> srcList = Arrays.asList(10, 11, 20, 21, 30, 31);        Observable.from(srcList)                .window(3)                .flatMap(o -> o.toList())                .subscribe(list -> log.info(list.toString()));    }    /**     * window 建立操纵符 建立滑动窗口     * 演示 window 建立操纵符   建立滑动窗口     */    @Test    public void windowObserverDemo()    {        List<Integer> srcList = Arrays.asList(10, 11, 20, 21, 30, 31);        Observable.from(srcList)                .window(3, 1)                .flatMap(o -> o.toList())                .subscribe(list -> log.info(list.toString()));    }    /**     * window 建立操纵符 建立滑动窗口     * 演示 window 滑动窗口和合并     */    @Test    public void windowMergeDemo()    {        List<Integer> srcList = Arrays.asList(10, 11, 20, 21, 30, 31);        Observable.merge(                Observable.from(srcList)                        .window(3, 1))                .subscribe(integer -> log.info(integer.toString()));    }    /**     * window 建立操纵符 建立时候窗口     * 演示  window 建立操纵符   建立时候窗口     */    @Test    public void timeWindowObserverDemo() throws InterruptedException    {        Observable eventStream = Observable                .interval(100, TimeUnit.MILLISECONDS);        eventStream.window(300, TimeUnit.MILLISECONDS)                .flatMap(o -> ((Observable<Integer>) o).toList())                .subscribe(list -> log.info(list.toString()));        Thread.sleep(Integer.MAX_VALUE);    }    /**     * 演示  hystrix 的  健康统计 metric     */    @Test    public void hystrixTimewindowDemo() throws InterruptedException    {        //建立Random类工具        Random random = new Random();        //模拟 Hystrix event 事务流，每 100ms 发送一个 0或1随机值， 随机值为 0 代表失利，机值为 1 代表成功        Observable eventStream = Observable                .interval(100, TimeUnit.MILLISECONDS)                .map(i -> random.nextInt(2));        /**         *完成桶内0值计数的聚合函数         */        Func1 reduceBucketToSummary = new Func1<Observable<Integer>, Observable<Long>>()        {            @Override            public Observable<Long> call(Observable<Integer> eventBucket)            {                Observable<List<Integer>> olist = eventBucket.toList();                Observable<Long> countValue = olist.map(list ->                {                    long count = list.stream().filter(i -> i == 0).count();                    log.info("{} '0 count:{}", list.toString(), count);                    return count;                });                return countValue;            }        };        /**         *桶计数流         */        Observable<Long> bucketedCounterStream = eventStream                .window(300, TimeUnit.MILLISECONDS)                .flatMap(reduceBucketToSummary);  // 将时候桶停止聚合，统计为事务值为0 的个数        /**         * 滑动窗口聚合函数         */        Func1 reduceWindowToSummary = new Func1<Observable<Long>, Observable<Long>>()        {            @Override            public Observable<Long> call(Observable<Long> eventBucket)            {                return eventBucket.reduce(new Func2<Long, Long, Long>()                {                    @Override                    public Long call(Long bucket1, Long bucket2)                    {                        /**                         * 对窗口内的桶，停止的累加                         */                        return bucket1 + bucket2;                    }                });            }        };        /**         * 桶滑动统计流         */        Observable bucketedRollingCounterStream = bucketedCounterStream                .window(3, 1)                .flatMap(reduceWindowToSummary);// 将滑动窗口停止聚合        bucketedRollingCounterStream.subscribe(sum -> log.info("滑动窗口的和：{}", sum));        Thread.sleep(Integer.MAX_VALUE);    }}