Soul 网关源码学习(18) - Soul 网关插件 Resilience4J

2021-02-04|字数总计: 1.5k|阅读时长: 5 分钟

soul 网关中关于熔断和限流的插件的可选实现除了有 HystrixPlugin, SentinelPlugin, 还有 Resilience4JPlugin。

Resilience4JPlugin 源码分析

什么是 Resilience4J

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

要使用Resilience4j，不需要引入所有依赖，只需要选择你需要的。

Resilience4j提供了以下的核心模块和拓展模块:

核心模块：

resilience4j-circuitbreaker: 断路器实现
resilience4j-ratelimiter: 限流实现
resilience4j-bulkhead: 隔离
resilience4j-retry: 自动重试 (同步和异步都支持)
resilience4j-cache: 结果缓存
resilience4j-timelimiter: 限时器实现

Soul 网关插件 Resilience4JPlugin 的实现模块 soul-plugin-resilience4j 中引入了依赖 resilience4j-circuitbreaker, resilience4j-ratelimiter 和 resilience4j-timelimiter 三个模块来实现
Resilience4JPlugin 对 Soul 网关中的请求的熔断和限流的处理。

源码分析

Resilience4JPlugin 通过继承 AbstractSoulPlugin 完成插件的实现，Resilience4JPlugin 的插件逻辑在方法 doExecute 中实现，Resilience4JPlugin 对象中定义了两个属性: org.dromara.soul.plugin.resilience4j.executor.CombinedExecutor
对象的实例 combinedExecutor 和 org.dromara.soul.plugin.resilience4j.executor.RateLimiterExecutor 对象的实例 ratelimiterExecutor, 在构造函数中初始化。

private final CombinedExecutor combinedExecutor;

private final RateLimiterExecutor ratelimiterExecutor;

public Resilience4JPlugin(final CombinedExecutor combinedExecutor,
                          final RateLimiterExecutor ratelimiterExecutor) {
    this.combinedExecutor = combinedExecutor;
    this.ratelimiterExecutor = ratelimiterExecutor;
}

org.dromara.soul.plugin.resilience4j.executor.CombinedExecutor 和 org.dromara.soul.plugin.resilience4j.executor.RateLimiterExecutor 是接口 org.dromara.soul.plugin. resilience4j.executor.Executor 的实现类, Executor 方法中定义了如下几个方法:

<T> Mono<T> run(Mono<T> toRun, Function<Throwable, Mono<T>> fallback, Resilience4JConf conf)

这个方法定义了 Executor 以什么逻辑 “执行”, 返回值是一个 Mono 对象。入参 toRun 为传入的响应式的执行单元; fallback 参数的类型是 Function<Throwable, Mono<T>>, 定义了通过异常如何转换出回退的响应式执行单元;
conf 参数的类型是 Resilience4JConf, 从调用上下文可以知道它是从 Resilience4JPlugin 的 doExecute 方法中入参的 rule 通过执行 org.dromara.soul.plugin.resilience4j.build.Resilience4JBuilder#build
方法后得到的, Resilience4JConf 对象是 Resilience4JPlugin 的配置信息的聚合, 它的定义如下:

@Data
public class Resilience4JConf {

    private String id;

    private String fallBackUri;

    private TimeLimiterConfig timeLimiterConfig;

    private CircuitBreakerConfig circuitBreakerConfig;

    private RateLimiterConfig rateLimiterConfig;

    public Resilience4JConf(final String id,
                            final String fallBackUri,
                            final RateLimiterConfig rateLimiterConfig,
                            final TimeLimiterConfig timeLimiterConfig,
                            final CircuitBreakerConfig circuitBreakerConfig) {
        this.id = id;
        this.fallBackUri = fallBackUri;
        this.rateLimiterConfig = rateLimiterConfig;
        this.timeLimiterConfig = timeLimiterConfig;
        this.circuitBreakerConfig = circuitBreakerConfig;
    }
}

id 属性定义了 Soul 网关中 Resilience4J 插件执行的运行空间, 因为一个 id 会对应一组 Resilience4J 运行时的配置信息。
fallBackUri 定义了经 Resilience4J 的熔断和降级发生了异常的请求的回退请求 URL。
timeLimiterConfig 的类型是 io.github.resilience4j.timelimiter.TimeLimiterConfig, 是依赖模块 resilience4j-timelimiter 定义的配置信息, 定义了 Resilience4J 限时器的运行时行为。
circuitBreakerConfig 的类型是 io.github.resilience4j.circuitbreaker.CircuitBreakerConfig 是依赖模块 resilience4j-circuitbreaker 定义的配置信息，定义了 Resilience4J 断路器的运行时行为。
rateLimiterConfig 的类型是 io.github.resilience4j.ratelimiter.RateLimiterConfig 是依赖模块 resilience4j-ratelimiter 定义的配置信息，定义了 Resilience4J 限流器的运行时行为。

Mono<Void> fallback(ServerWebExchange exchange, String uri, Throwable t)

这是接口中的 default 方法, 定义了如何执行回退请求 URL, 这里需要注意的是, 最终执行 fallbackUri 请求的是 DispatcherHandler 类型 Bean 的实例, 这个 Bean 是在 org.dromara.soul.web.configuration.SoulConfiguration 中向
Spring 容器注册的。

Mono<Void> withoutFallback(ServerWebExchange exchange, Throwable throwable)

这个方法也是接口中的 default 方法, 定义了如果回退请求 URL不存在的时候, soul 网关如何将熔断、限流或请求超时的异常信息返回个请求的客户端。

org.dromara.soul.plugin.resilience4j.executor.RateLimiterExecutor 接口定义了 Resilience4JPlugin 如何将 Resilience4J 的限流功能”植入”到网关响应式的执行流水线中,
org.dromara.soul.plugin.resilience4j.executor.CombinedExecutor 接口定义了 Resilience4JPlugin 如何将 Resilience4J 的断路器, 限流和限时请求调用”植入”到网关响应式的执行流水线中。Resilience4JPlugin 的 doExecute
方法的逻辑就非常清晰了:

    @Override
    protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
        final SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        assert soulContext != null;
        Resilience4JHandle resilience4JHandle = GsonUtils.getGson().fromJson(rule.getHandle(), Resilience4JHandle.class);
        if (resilience4JHandle.getCircuitEnable() == 1) {
            return combined(exchange, chain, rule);
        }
        return rateLimiter(exchange, chain, rule);
    }

    private Mono<Void> rateLimiter(final ServerWebExchange exchange, final SoulPluginChain chain, final RuleData rule) {
        return ratelimiterExecutor.run(
                chain.execute(exchange), fallback(ratelimiterExecutor, exchange, null), Resilience4JBuilder.build(rule))
                .onErrorResume(throwable -> ratelimiterExecutor.withoutFallback(exchange, throwable));
    }

    private Mono<Void> combined(final ServerWebExchange exchange, final SoulPluginChain chain, final RuleData rule) {
        Resilience4JConf conf = Resilience4JBuilder.build(rule);
        return combinedExecutor.run(
                chain.execute(exchange).doOnSuccess(v -> {
                    HttpStatus status = exchange.getResponse().getStatusCode();
                    if (status == null || !status.is2xxSuccessful()) {
                        exchange.getResponse().setStatusCode(null);
                        throw new CircuitBreakerStatusCodeException(status == null ? HttpStatus.INTERNAL_SERVER_ERROR : status);
                    }
                }), fallback(combinedExecutor, exchange, conf.getFallBackUri()), conf);
    }

    private Function<Throwable, Mono<Void>> fallback(final Executor executor,
                                                     final ServerWebExchange exchange, final String uri) {
        return throwable -> executor.fallback(exchange, uri, throwable);
    }
}

从传入的 rule 信息中获取 PluginHandle 信息, 构造 Resilience4JHandle 对象, Resilience4JHandle 中定义了 Resilience4J 运行时的配置。
判断 Resilience4JHandle 信息中是否开启了断路器功能:
- 如果开启了，插件最终通过 combinedExecutor 完成将 Resilience4J 的断路器, 限流和限时请求调用”植入”到网关响应式的执行流水线中。
- 如果未开启, 插件最终通过 ratelimiterExecutor 完成将 Resilience4J 的限流功能”植入”到网关响应式的执行流水线中。

总结

通过今天的源码学习, 了解到了一个新的可以实现熔断, 限流和请求限时调用的框架 Resilience4J, 从这个框架的简介中了解到了它的强大。同时在 Resilience4JPlugin 的源码分析的过程中看到了这个插件实现的优雅的地方,
其中不乏优秀的编码方式和设计模式合理地使用, 让人觉得看得不过瘾~~

参考

https://www.jianshu.com/p/5531b66b777a