专栏首页码匠的流水账聊聊sentinel的SentinelGatewayFilter

聊聊sentinel的SentinelGatewayFilter

本文主要研究一下sentinel的SentinelGatewayFilter

SentinelGatewayFilter

Sentinel-1.6.2/sentinel-adapter/sentinel-spring-cloud-gateway-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/gateway/sc/SentinelGatewayFilter.java

public class SentinelGatewayFilter implements GatewayFilter, GlobalFilter {

    private final GatewayParamParser<ServerWebExchange> paramParser = new GatewayParamParser<>(
        new ServerWebExchangeItemParser());

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        Route route = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);

        Mono<Void> asyncResult = chain.filter(exchange);
        if (route != null) {
            String routeId = route.getId();
            Object[] params = paramParser.parseParameterFor(routeId, exchange,
                r -> r.getResourceMode() == SentinelGatewayConstants.RESOURCE_MODE_ROUTE_ID);
            String origin = Optional.ofNullable(GatewayCallbackManager.getRequestOriginParser())
                .map(f -> f.apply(exchange))
                .orElse("");
            asyncResult = asyncResult.transform(
                new SentinelReactorTransformer<>(new EntryConfig(routeId, EntryType.IN,
                    1, params, new ContextConfig(contextName(routeId), origin)))
            );
        }

        Set<String> matchingApis = pickMatchingApiDefinitions(exchange);
        for (String apiName : matchingApis) {
            Object[] params = paramParser.parseParameterFor(apiName, exchange,
                r -> r.getResourceMode() == SentinelGatewayConstants.RESOURCE_MODE_CUSTOM_API_NAME);
            asyncResult = asyncResult.transform(
                new SentinelReactorTransformer<>(new EntryConfig(apiName, EntryType.IN, 1, params))
            );
        }

        return asyncResult;
    }

    private String contextName(String route) {
        return SentinelGatewayConstants.GATEWAY_CONTEXT_ROUTE_PREFIX + route;
    }

    Set<String> pickMatchingApiDefinitions(ServerWebExchange exchange) {
        return GatewayApiMatcherManager.getApiMatcherMap().values()
            .stream()
            .filter(m -> m.test(exchange))
            .map(WebExchangeApiMatcher::getApiName)
            .collect(Collectors.toSet());
    }
}
  • SentinelGatewayFilter实现了GatewayFilter、GlobalFilter接口;其filter方法主要是获取route信息,然后对asyncResult进行transform,这里使用的是SentinelReactorTransformer

SentinelReactorTransformer

Sentinel-1.6.2/sentinel-adapter/sentinel-reactor-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/reactor/SentinelReactorTransformer.java

public class SentinelReactorTransformer<T> implements Function<Publisher<T>, Publisher<T>> {

    private final EntryConfig entryConfig;

    public SentinelReactorTransformer(String resourceName) {
        this(new EntryConfig(resourceName));
    }

    public SentinelReactorTransformer(EntryConfig entryConfig) {
        AssertUtil.notNull(entryConfig, "entryConfig cannot be null");
        this.entryConfig = entryConfig;
    }

    @Override
    public Publisher<T> apply(Publisher<T> publisher) {
        if (publisher instanceof Mono) {
            return new MonoSentinelOperator<>((Mono<T>) publisher, entryConfig);
        }
        if (publisher instanceof Flux) {
            return new FluxSentinelOperator<>((Flux<T>) publisher, entryConfig);
        }

        throw new IllegalStateException("Publisher type is not supported: " + publisher.getClass().getCanonicalName());
    }
}
  • SentinelReactorTransformer使用entryConfig创建了MonoSentinelOperator或者MonoSentinelOperator

MonoSentinelOperator

Sentinel-1.6.2/sentinel-adapter/sentinel-reactor-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/reactor/MonoSentinelOperator.java

public class MonoSentinelOperator<T> extends MonoOperator<T, T> {

    private final EntryConfig entryConfig;

    public MonoSentinelOperator(Mono<? extends T> source, EntryConfig entryConfig) {
        super(source);
        AssertUtil.notNull(entryConfig, "entryConfig cannot be null");
        this.entryConfig = entryConfig;
    }

    @Override
    public void subscribe(CoreSubscriber<? super T> actual) {
        source.subscribe(new SentinelReactorSubscriber<>(entryConfig, actual, true));
    }
}
  • MonoSentinelOperator在subscribe的时候,使用的是SentinelReactorSubscriber

FluxSentinelOperator

Sentinel-1.6.2/sentinel-adapter/sentinel-reactor-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/reactor/FluxSentinelOperator.java

public class FluxSentinelOperator<T> extends FluxOperator<T, T> {

    private final EntryConfig entryConfig;

    public FluxSentinelOperator(Flux<? extends T> source, EntryConfig entryConfig) {
        super(source);
        AssertUtil.notNull(entryConfig, "entryConfig cannot be null");
        this.entryConfig = entryConfig;
    }

    @Override
    public void subscribe(CoreSubscriber<? super T> actual) {
        source.subscribe(new SentinelReactorSubscriber<>(entryConfig, actual, false));
    }
}
  • FluxSentinelOperator在subscribe的时候,使用的是SentinelReactorSubscriber

SentinelReactorSubscriber

Sentinel-1.6.2/sentinel-adapter/sentinel-reactor-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/reactor/SentinelReactorSubscriber.java

public class SentinelReactorSubscriber<T> extends InheritableBaseSubscriber<T> {

    private final EntryConfig entryConfig;

    private final CoreSubscriber<? super T> actual;
    private final boolean unary;

    private volatile AsyncEntry currentEntry;
    private final AtomicBoolean entryExited = new AtomicBoolean(false);

    public SentinelReactorSubscriber(EntryConfig entryConfig,
                                     CoreSubscriber<? super T> actual,
                                     boolean unary) {
        checkEntryConfig(entryConfig);
        this.entryConfig = entryConfig;
        this.actual = actual;
        this.unary = unary;
    }

    private void checkEntryConfig(EntryConfig config) {
        AssertUtil.notNull(config, "entryConfig cannot be null");
    }

    @Override
    public Context currentContext() {
        if (currentEntry == null || entryExited.get()) {
            return actual.currentContext();
        }
        com.alibaba.csp.sentinel.context.Context sentinelContext = currentEntry.getAsyncContext();
        if (sentinelContext == null) {
            return actual.currentContext();
        }
        return actual.currentContext()
            .put(SentinelReactorConstants.SENTINEL_CONTEXT_KEY, currentEntry.getAsyncContext());
    }

    private void doWithContextOrCurrent(Supplier<Optional<com.alibaba.csp.sentinel.context.Context>> contextSupplier,
                                        Runnable f) {
        Optional<com.alibaba.csp.sentinel.context.Context> contextOpt = contextSupplier.get();
        if (!contextOpt.isPresent()) {
            // Provided context is absent, use current context.
            f.run();
        } else {
            // Run on provided context.
            ContextUtil.runOnContext(contextOpt.get(), f);
        }
    }

    private void entryWhenSubscribed() {
        ContextConfig sentinelContextConfig = entryConfig.getContextConfig();
        if (sentinelContextConfig != null) {
            // If current we're already in a context, the context config won't work.
            ContextUtil.enter(sentinelContextConfig.getContextName(), sentinelContextConfig.getOrigin());
        }
        try {
            AsyncEntry entry = SphU.asyncEntry(entryConfig.getResourceName(), entryConfig.getEntryType(),
                entryConfig.getAcquireCount(), entryConfig.getArgs());
            this.currentEntry = entry;
            actual.onSubscribe(this);
        } catch (BlockException ex) {
            // Mark as completed (exited) explicitly.
            entryExited.set(true);
            // Signal cancel and propagate the {@code BlockException}.
            cancel();
            actual.onSubscribe(this);
            actual.onError(ex);
        } finally {
            if (sentinelContextConfig != null) {
                ContextUtil.exit();
            }
        }
    }

    @Override
    protected void hookOnSubscribe(Subscription subscription) {
        doWithContextOrCurrent(() -> currentContext().getOrEmpty(SentinelReactorConstants.SENTINEL_CONTEXT_KEY),
            this::entryWhenSubscribed);
    }

    @Override
    protected void hookOnNext(T value) {
        if (isDisposed()) {
            tryCompleteEntry();
            return;
        }
        doWithContextOrCurrent(() -> Optional.ofNullable(currentEntry).map(AsyncEntry::getAsyncContext),
            () -> actual.onNext(value));

        if (unary) {
            // For some cases of unary operator (Mono), we have to do this during onNext hook.
            // e.g. this kind of order: onSubscribe() -> onNext() -> cancel() -> onComplete()
            // the onComplete hook will not be executed so we'll need to complete the entry in advance.
            tryCompleteEntry();
        }
    }

    @Override
    protected void hookOnComplete() {
        tryCompleteEntry();
        actual.onComplete();
    }

    @Override
    protected boolean shouldCallErrorDropHook() {
        // When flow control triggered or stream terminated, the incoming
        // deprecated exceptions should be dropped implicitly, so we'll not call the `onErrorDropped` hook.
        return !entryExited.get();
    }

    @Override
    protected void hookOnError(Throwable t) {
        if (currentEntry != null && currentEntry.getAsyncContext() != null) {
            // Normal requests with non-BlockException will go through here.
            Tracer.traceContext(t, 1, currentEntry.getAsyncContext());
        }
        tryCompleteEntry();
        actual.onError(t);
    }

    @Override
    protected void hookOnCancel() {

    }

    private boolean tryCompleteEntry() {
        if (currentEntry != null && entryExited.compareAndSet(false, true)) {
            currentEntry.exit(1, entryConfig.getArgs());
            return true;
        }
        return false;
    }
}
  • SentinelReactorSubscriber继承了InheritableBaseSubscriber(拷贝自reactor.core.publisher.BaseSubscriber,允许子类覆盖onSubscribe、onNext、onError、onComplete方法)
  • 这里hookOnSubscribe调用了entryWhenSubscribed,它在sentinelContextConfig不为null的时候会先执行ContextUtil.enter,然后使用SphU.asyncEntry创建了AsyncEntry,最后在finally里头在sentinelContextConfig不为null的时候执行ContextUtil.exit();
  • 这里hookOnNext、hookOnComplete、hookOnError都调用了tryCompleteEntry方法,它主要是尝试退出AsyncEntry

小结

  • SentinelGatewayFilter实现了GatewayFilter、GlobalFilter接口;其filter方法主要是获取route信息,然后对asyncResult进行transform,这里使用的是SentinelReactorTransformer
  • SentinelReactorTransformer使用entryConfig创建了MonoSentinelOperator或者MonoSentinelOperator;它们在subscribe的时候,使用的是SentinelReactorSubscriber
  • SentinelReactorSubscriber主要是在hookOnSubscribe的时候调用了entryWhenSubscribed方法创建AsyncEntry,在hookOnNext、hookOnComplete、hookOnError的时候调用了tryCompleteEntry方法,尝试退出AsyncEntry

doc

  • SentinelGatewayFilter

本文分享自微信公众号 - 码匠的流水账(geek_luandun),作者:码匠乱炖

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-07-07

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 聊聊sentinel的SentinelGatewayFilter

    Sentinel-1.6.2/sentinel-adapter/sentinel-spring-cloud-gateway-adapter/src/main/j...

    codecraft
  • 聊聊scalecube-cluster的GossipProtocol

    scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/gossip/Gossip...

    codecraft
  • 聊聊scalecube-cluster的GossipProtocol

    scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/gossip/Gossip...

    codecraft
  • 聊聊sentinel的SentinelGatewayFilter

    Sentinel-1.6.2/sentinel-adapter/sentinel-spring-cloud-gateway-adapter/src/main/j...

    codecraft
  • 从锅炉工到AI专家(4)

    俺踏月色而来
  • 迪斯尼利用人工智能实现声音与图片之间的关联

    美国迪斯尼研究中心发报告称,其与瑞士苏黎世联邦理工学院合作利用人工智能设计出了能自主匹配图像和声音的系统。 儿童可以根据不同动物的声音从图书中学习关联图像和声音...

    人工智能快报
  • 如何创建SAP Cloud Platform Process Integration runtime服务

    登录SAP Cloud Platform cockpit,进入subaccount的Service marketplace界面,选择process integr...

    Jerry Wang
  • android入门 — Service

    Mister24
  • SharePoint 关于拓扑错误的解决方案

    Issue Topology报错信息:SharePoint Web Services Round Robin Service Load Balancer Eve...

    用户1161731
  • 使用ABAP绘制可伸缩矢量图

    Jerry去年的文章 动手使用ABAP Channel开发一些小工具,提升日常工作效率 里曾经介绍过一些用ABAP实现的可供娱乐的小程序,比如用古老的HPGL接...

    Jerry Wang

扫码关注云+社区

领取腾讯云代金券