原理

基于 DubboFilter 机制，实现 SentinelFilter，在其中定义 Sentinel 的模板代码即可。provider 和 consumer 均提供了扩展。

image.png

对于 app2 来说，既是 app1 的 provider，也是 app3 的 consumer。
首先 consumer 使用 DubboAppContextFilter 将应用来源信息填充到上下文，consumer 使用 SentinelDubboConsumerFilter 进行 out 级别的限流保护，
provider 使用 SentinelDubboProviderFilter 进行 in 级别的限流保护（并且填充来源为上下文中的来源）。

SPI 配置文件

META-INF/dubbo/org.apache.dubbo.rpc.Filter：三个 Filter 扩展

sentinel.dubbo.provider.filter=com.alibaba.csp.sentinel.adapter.dubbo.SentinelDubboProviderFilter
sentinel.dubbo.consumer.filter=com.alibaba.csp.sentinel.adapter.dubbo.SentinelDubboConsumerFilter
dubbo.application.context.name.filter=com.alibaba.csp.sentinel.adapter.dubbo.DubboAppContextFilter

填充来源 Filter

@Activate(group = CONSUMER)
public class DubboAppContextFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String application = invoker.getUrl().getParameter(CommonConstants.APPLICATION_KEY);
        if (application != null) {
            // 将 application 值设置到上下文
            RpcContext.getContext().setAttachment(DubboUtils.SENTINEL_DUBBO_APPLICATION_KEY, application);
        }
        return invoker.invoke(invocation);
    }
}

Provider Filter

@Activate(group = PROVIDER)
public class SentinelDubboProviderFilter extends BaseSentinelDubboFilter {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        // 从上下文获取 application 作为调用来源
        String origin = DubboAdapterGlobalConfig.getOriginParser().parse(invoker, invocation);
        ...
        Entry interfaceEntry = null;
        Entry methodEntry = null;
        ...
        try {
            ContextUtil.enter(methodResourceName, origin);
            // 创建接口级别和方法级别的两个 Entry
            interfaceEntry = SphU.entry(interfaceResourceName, ResourceTypeConstants.COMMON_RPC, EntryType.IN);
            methodEntry = SphU.entry(methodResourceName, ResourceTypeConstants.COMMON_RPC, EntryType.IN, invocation.getArguments());
            // 业务逻辑
            Result result = invoker.invoke(invocation);
            // 失败异常记录，用于异常熔断降级
            if (result.hasException()) {
                Tracer.traceEntry(result.getException(), interfaceEntry);
                Tracer.traceEntry(result.getException(), methodEntry);
            }
            return result;
        } catch (BlockException e) {
            // 使用全局流控回调器
            return DubboAdapterGlobalConfig.getProviderFallback().handle(invoker, invocation, e);
        } catch (RpcException e) {
            // 失败异常记录，用于异常熔断降级
            Tracer.traceEntry(e, interfaceEntry);
            Tracer.traceEntry(e, methodEntry);
            throw e;
        } finally {
            // exit
            if (methodEntry != null) {
                methodEntry.exit(1, invocation.getArguments());
            }
            if (interfaceEntry != null) {
                interfaceEntry.exit();
            }
            ContextUtil.exit();
        }
    }
}

Consumer Filter

@Activate(group = CONSUMER)
public class SentinelDubboConsumerFilter extends BaseSentinelDubboFilter {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        InvokeMode invokeMode = RpcUtils.getInvokeMode(invoker.getUrl(), invocation);
        if (InvokeMode.SYNC == invokeMode) {
            // 同步模式，与provider一致，只是EntryType为out；另外就是流控异常全局处理器为 consumerFallback
            return syncInvoke(invoker, invocation);
        } else {
            // 异步模式
            return asyncInvoke(invoker, invocation);
        }
    }

    private Result syncInvoke(Invoker<?> invoker, Invocation invocation) {
        ...
    }

    private Result asyncInvoke(Invoker<?> invoker, Invocation invocation) {
        LinkedList<EntryHolder> queue = new LinkedList<>();
        ...
        try {
            // 获取 Entry
            queue.push(new EntryHolder(SphU.asyncEntry(interfaceResourceName, ResourceTypeConstants.COMMON_RPC, EntryType.OUT), null));
            queue.push(new EntryHolder(SphU.asyncEntry(methodResourceName, ResourceTypeConstants.COMMON_RPC, EntryType.OUT, 1, invocation.getArguments()), invocation.getArguments()));
            // 业务逻辑
            Result result = invoker.invoke(invocation);
            // 异步回调
            result.whenCompleteWithContext((r, throwable) -> {
                Throwable error = throwable;
                if (error == null) {
                    error = Optional.ofNullable(r).map(Result::getException).orElse(null);
                }
                while (!queue.isEmpty()) {
                    EntryHolder holder = queue.pop();
                    // 记录异常，用于熔断降级（Tracer.traceEntry 内部判断了error为null）
                    Tracer.traceEntry(error, holder.entry);
                    // 退出 holder.entry.exit() 或 holder.entry.exit(1, holder.params)
                    exitEntry(holder);
                }
            });
            return result;
        } catch (BlockException e) {
            while (!queue.isEmpty()) {
                exitEntry(queue.pop());
            }
            return DubboAdapterGlobalConfig.getConsumerFallback().handle(invoker, invocation, e);
        }
    }

流控异常全局处理器

@FunctionalInterface
public interface DubboFallback {

    /**
     * Handle the block exception and provide fallback result.
     *
     * @param invoker Dubbo invoker
     * @param invocation Dubbo invocation
     * @param ex block exception
     * @return fallback result
     */
    Result handle(Invoker<?> invoker, Invocation invocation, BlockException ex);
}

DubboAdapterGlobalConfig 中默认提供了两个实现。

private static volatile DubboFallback consumerFallback = new DefaultDubboFallback();
private static volatile DubboFallback providerFallback = new DefaultDubboFallback();

应用可通过 DubboAdapterGlobalConfig.setConsumerFallback(DubboFallback) 和 DubboAdapterGlobalConfig.setProviderFallback(DubboFallback) 来覆盖。