Circuit Breaker Back Pressuire

不少同学在学习Reactive的时候，突然多了一个Back Pressure(背压)的概念，而大家之前接触的更多的是Circuit Breaker(断路保护)，这篇文章就说明一下Circuit Breaker和Back Pressure之间的区别，以及如何在Reactive模式中接入断路保护和背压。

应用间调用

假设系统中有App-1要调用App-2的服务，如REST API，HSF或者Dubbo等，都没有问题。 如果App-1发起了非常多的请求，如程序的Bug、突发流量(秒杀)或者被攻击等， 这个时候App-2就会收到非常多的请求，而App-2自身无法在短时间内处理这些请求，所以要拒绝一部分请求，这个也就是我们常说的限流。 如果请求量实在太大，那就是DDoS攻击，App-2几乎无法处理任何请求，大部分计算能力都忙于计算丢弃这个任务。 DDoS这个不表，我们接下来主要说一下限流。

断路保护之限流

当系统无法处理所有的请求时，自然要丢弃一部分请求，还能让部分请求能被正常处理。 这里就有一个RateLimiter的概念， 你理解为速度限制，如每秒钟发起的请求数量，当前带宽的容量，CPU负载等，也就是超过这个限制，就会触发限流。 为了方便接下来的解释，我们将限制假定为QPS，也就是每秒钟的请求数，这个也是比较典型的场景，如RPC调用等。

接收方限流

这个也是大家都比较容易理解的，我接收到的请求太多啦，超出了系统的限制，当然我会丢弃一部分请求。 接收方丢弃一个请求成本还是比较高的，首先这个请求已经在请求方组装好啦，而且通过网络发送到接收方，而现在被丢弃掉，这等于之前所有的工作都是无用功，计算都被浪费， 同时失败的时间被拉长，不是快速失败。 另外对请求方来说，还要有对应的丢弃逻辑和运行期计算。 对于Java程序员来说，你可以理解为一个Servlet Filter，你需要设计和实现该Filter，同时该Filter每次都要参与到运行器计算中，也增加了不少成本。

发起方限流

还有一种是在请求方进行限流，请求发起方会根据一定的参数计算出一个限制值，然后根据此限制值判断是否要将请求发送到服务端。 对比接收方限流，发起方限流更高效且更节省资源。 所以我们更希望的是在请求发起端进行限流，不要发送无谓的请求到服务端造成浪费。 但是客户端限流也有对应的工作量，你需要设计反馈机制，不然发起方无从得知接收方的系统情况，我们设计一套反馈机制并提供相应的SDK，从而实现限流反馈。

Reactive背压

在实际应用中，还有一种数据处理的场景，如订阅某一消息系统的Topic，然后从该Topic接收消息进行相应的逻辑处理。 这里也有两个模式，一种是消息推送，一种是消息轮询，接下我们就看一下如何处理这种流式消息场景。

消息推送

这个大家都能明白，消息系统在收到消息后，然后转发给消息消费方，非常高效和实时。但是这里有一个问题， 如果瞬间大量的消息推送到消息消费方，消费方无法处理这么多的请求，这个时候怎么办？ 同样也会出现限流，导致一部分消息没有被处理。对于消息推送模式来说， 消息消费方完全出于无保护状态，相对风险比较高。 传统的消息中间件都采用这个架构。

消息轮询

另外一种是消息轮询，也就是消息消费方主动向消息系统请求消息，如每次100条，消费方有多大能力，就可以请求多少消息，这样消息消费方就可以进行自我保护。 但是这里也有一些问题，消费方每次都要轮询，可能发起了请求却没有数据消息，出现轮空的情况，这也是为何大家抱怨Ajax Polling效率不高，更多人原因选择的WebSocket的原因。 同时发起方还要维护轮询的时间间隔和位点等信息，需要借助ZooKeeper等系统进行分布式协作，增加系统复杂度，如Kafka就是这个模式。

背压

接下来我们看一下背压是如何工作的。 首先消息消费方告诉消息系统，请给我100条数据，消息系统然后维护一个计数器，消息系统在接收到消息后，马上会转发给消息消费方，但是消息到了100条后，消息系统就不会再推送啦。 消息消费方在处理完所有的消息后，会再发起一个Request N的消息请求，然后消息系统会再推送N条消息，这个就是背压的流程。 介入背压后，消息消费方得到保护，不会出现过载推送这种情况，同时消息消费方也不要维护时间间隔和位点等信息，只要告诉消息系统给我N个消息就可以啦，这些信息都是由消息系统进行维护的，消费方简单了非常多，同时利用推送模型，实时性和效率都高了很多。

Reactive & RSocket如何解决限流和被压

接下来我们就看一下在Reactive和RSocket中如何实现限流和背压的，以及其背后的原理。

RSocket Stream被压

Reactive默认就提供被压支持，你只需要调用一下limitRate(int prefetchRate) API就可以，而RSocket完全实现了Reactive语义，其通讯协议集成了背压支持的REQUEST_N通讯指令，所以你只需要调用API就可以啦。如下:

rsocket.requestStream(payload)
                .limitRate(100)
                .subscribe(payload1 -> {

                })

如果你没有设置limitRate，则是Integer.MAX_VALUE，表示最多推送Integer.MAX_VALUE个消息。

RSocket RPC 限流

RSocket提供了RPC通讯模型，对比限流的模型，我们觉得请求方限流是非常好的策略。 假设我们设置应用和RSocket Broker直接RPC调用的QPS为10K，我们使用以下代码即可。

RateLimiterConfig config = RateLimiterConfig.custom()
                .limitRefreshPeriod(Duration.ofMillis(1000))
                .limitForPeriod(100000)
                .build();
RateLimiter rateLimiter = RateLimiter.of("rpcLimiter", config);
return rsocket.requestResponse(payload).transformDeferred(RateLimiterOperator.of(rateLimiter));

当然这个10K是可以动态调整的，如Broker可以通过metadataPush推送RateLimiterConfigEvent事件，这样就可以动态调整响应的限制。

如果QPS超过10K，那么RateLimiter就会发生作用，多余的请求就会被丢弃，不会发送给服务接收方，造成服务端压力。 多余的请求被快速丢弃，这个也实现了fast fail(快速失败)的设计需求。

有同学会问，能否将RPC调用转换为Stream，然后以背压的方式来实现限流？ 这个是可以做到的，但是比较复杂。

• 首先RPC调用对RT要求比较高，这不同于数据处理，所以必须要快，要不快速成功，要不快速失败；
• 另外RPC并没有被压的概念，你需要将RPC请求调整到Stream上，然后再走Stream的接口，这个处理起来比较复杂，RSocket并没有对应的支持。
• 如果RPC转换为Stream后，还需要处理单个RPC请求超时、如何取消等问题，这些也都非常麻烦。

综合下来，个人建议还是RPC采用限流方案，让后使用RSocket的metadataPush进行实时限流反馈，这个事件在RSocket Broker也默认提供。