一、从一次 HTTP 请求开始

在一个生产环境中，服务节点通常暴露了成百上千个 HTTP 接口对外提供服务。为了保证系统的稳定性，核心 HTTP 接口往往需要配置限流规则。给 HTTP 接口配置限流，可以防止突发或恶意的高并发请求耗尽服务器资源（如 CPU、内存、数据库连接等），从而避免服务崩溃或引发雪崩效应。

基础示例

假设我们有下面这样一个 HTTP 接口，需要给它配置限流规则：

@RestController
@RequiredArgsConstructor
@RequestMapping("/demo")
public class DemoController {

    @RequestMapping("/hello")
    @SentinelResource("test_sentinel")
    public String hello() {
        return "hello world";
    }
}

使用起来非常简单。首先我们可以选择给接口加上 @SentinelResource 注解（也可以不加，如果不加 Sentinel 客户端会使用请求路径作为资源名，详细原理在后面章节讲解），然后到流控控制台给该资源配置流控规则即可。

二、限流规则的加载

限流规则的生效，是从限流规则的加载开始的。聚焦到客户端的 RuleLoader 类，可以看到它支持了多种规则的加载：

• 流控规则；
• 集群限流规则；
• 熔断规则；
• ......

RuleLoader 核心逻辑

RuleLoader 类的核心作用是将这些规则加载到缓存中，方便后续使用：

public class RuleLoader {

    /**
     * 加载所有 Sentinel 规则到内存缓存
     *
     * @param sentinelRules 包含各种规则的配置对象
     */
    public static void loadRule(SentinelRules sentinelRules) {
        if (sentinelRules == null) {
            return;
        }

        // 加载流控规则
        FlowRuleManager.loadRules(sentinelRules.getFlowRules());
        // 加载集群流控规则
        RuleManager.loadClusterFlowRule(sentinelRules.getFlowRules());

        // 加载参数流控规则
        ParamFlowRuleManager.loadRules(sentinelRules.getParamFlowRules());
        // 加载参数集群流控规则
        RuleManager.loadClusterParamFlowRule(sentinelRules.getParamFlowRules());

        // 加载熔断规则
        DegradeRuleManager.loadRules(sentinelRules.getDegradeRules());

        // 加载参数熔断规则
        ParamDegradeRuleManager.loadRules(sentinelRules.getParamDegradeRules());

        // 加载系统限流规则
        SystemRuleManager.loadRules(sentinelRules.getSystemRules());
    }
}

流控规则加载详情

以流控规则的加载为例深入FlowRuleManager.loadRules 方法可以看到其完整的加载逻辑：

public static void loadRules(List<FlowRule> rules) {
    // 通过动态配置属性更新规则值
    currentProperty.updateValue(rules);
}

updateValue 方法负责通知所有监听器配置变更：

public boolean updateValue(T newValue) {
    // 如果新旧值相同，无需更新
    if (isEqual(value, newValue)) {
        return false;
    }
    RecordLog.info("[DynamicSentinelProperty] Config will be updated to: " + newValue);

    // 更新配置值
    value = newValue;
    // 通知所有监听器配置已更新
    for (PropertyListener<T> listener : listeners) {
        listener.configUpdate(newValue);
    }
    return true;
}

FlowPropertyListener 是流控规则变更的具体监听器实现：

private static final class FlowPropertyListener implements PropertyListener<List<FlowRule>> {

    @Override
    public void configUpdate(List<FlowRule> value) {
        // 构建流控规则映射表（按资源名分组）
        Map<String, List<FlowRule>> rules = FlowRuleUtil.buildFlowRuleMap(value);
        if (rules != null) {
            // 清空旧规则
            flowRules.clear();
            // 加载新规则
            flowRules.putAll(rules);
        }
        RecordLog.info("[FlowRuleManager] Flow rules received: " + flowRules);
    }
}

三、SentinelServletFilter 过滤器

在 Sentinel 中，所有的资源都对应一个资源名称和一个 Entry。Entry 可以通过对主流框架的适配自动创建，也可以通过注解的方式或调用 API 显式创建。Entry 是限流的入口类，通过 @SentinelResource 注解的限流本质上也是通过 AOP 的方式进行了对 Entry 类的调用。

Entry 的编程范式

Entry 类的标准使用方式如下：

// 资源名可使用任意有业务语义的字符串，比如方法名、接口名或其它可唯一标识的字符串
try (Entry entry = SphU.entry("resourceName")) {
    // 被保护的业务逻辑
    // do something here...
} catch (BlockException ex) {
    // 资源访问阻止，被限流或被降级
    // 在此处进行相应的处理操作
}

Servlet Filter 拦截逻辑

对于一个 HTTP 资源，在没有显式标注 @SentinelResource 注解的情况下，会有一个 Servlet Filter 类 SentinelServletFilter 统一进行拦截：

public class SentinelServletFilter implements Filter {

    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
            throws IOException, ServletException {
        HttpServletRequest sRequest = (HttpServletRequest) request;
        Entry urlEntry = null;

        try {
            // 获取并清理请求路径
            String target = FilterUtil.filterTarget(sRequest);

            // 统一 URL 清理逻辑
            // 对于 RESTful API，必须对 URL 进行清理（例如将 /foo/1 和 /foo/2 统一为 /foo/:id），
            // 否则上下文和资源的数量会超过阈值
            SentinelUrlCleaner urlCleaner = SentinelUrlCleaner.SENTINEL_URL_CLEANER;
            if (urlCleaner != null) {
                target = urlCleaner.clean(sRequest, target);
            }

            // 如果请求路径不为空且非安全扫描，则进入限流逻辑
            if (!StringUtil.isEmpty(target) && !isSecScan) {
                // 解析来源标识（用于来源限流）
                String origin = parseOrigin(sRequest);
                // 确定上下文名称
                String contextName = webContextUnify
                    ? WebServletConfig.WEB_SERVLET_CONTEXT_NAME
                    : target;

                // 使用 WEB_SERVLET_CONTEXT_NAME 作为当前 Context 的名字
                ContextUtil.enter(contextName, origin);

                // 根据配置决定是否包含 HTTP 方法
                if (httpMethodSpecify) {
                    String pathWithHttpMethod = sRequest.getMethod().toUpperCase() + COLON + target;
                    // 实际进入到限流统计判断逻辑，资源名是 "方法:路径"
                    urlEntry = SphU.entry(pathWithHttpMethod, ResourceTypeConstants.COMMON_WEB, EntryType.IN);
                } else {
                    // 实际进入到限流统计判断逻辑，资源名是请求路径
                    urlEntry = SphU.entry(target, ResourceTypeConstants.COMMON_WEB, EntryType.IN);
                }
            }

            // 继续执行后续过滤器
            chain.doFilter(request, response);

        } catch (BlockException e) {
            // 处理被限流的情况
            HttpServletResponse sResponse = (HttpServletResponse) response;
            // 返回限流页面或重定向到其他 URL
            WebCallbackManager.getUrlBlockHandler().blocked(sRequest, sResponse, e);

        } catch (IOException | ServletException | RuntimeException e2) {
            // 记录异常信息用于统计
            Tracer.traceEntry(e2, urlEntry);
            throw e2;

        } finally {
            // 释放 Entry 资源
            if (urlEntry != null) {
                urlEntry.exit();
            }
            // 退出当前上下文
            ContextUtil.exit();
        }
    }
}

四、SentinelResourceAspect 切面

如果在接口上标注了 @SentinelResource 注解，还会有另外的逻辑处理。Sentinel 定义了一个单独的 AOP 切面 SentinelResourceAspect 专门用于处理注解限流。

SentinelResource 注解定义

先来看看 @SentinelResource 注解的完整定义：

@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface SentinelResource {

    /**
     * Sentinel 资源的名称（即资源标识）
     * 必填项，不能为空
     */
    String value() default "";

    /**
     * 资源的入口类型（入站 IN 或出站 OUT）
     * 默认为出站（OUT）
     */
    EntryType entryType() default EntryType.OUT;

    /**
     * 资源的分类（类型）
     * 自 1.7.0 版本起支持
     */
    int resourceType() default 0;

    /**
     * 限流或熔断时调用的 block 异常处理方法的名称
     * 默认为空（即不指定）
     */
    String blockHandler() default "";

    /**
     * blockHandler 所在的类
     * 如果与原方法不在同一个类，需要指定此参数
     */
    Class<?>[] blockHandlerClass() default {};

    /**
     * 降级（fallback）方法的名称
     * 默认为空（即不指定）
     */
    String fallback() default "";

    /**
     * 用作通用的默认降级方法
     * 该方法不能接收任何参数，且返回类型需与原方法兼容
     */
    String defaultFallback() default "";

    /**
     * fallback 所在的类
     * 如果与原方法不在同一个类，需要指定此参数
     */
    Class<?>[] fallbackClass() default {};

    /**
     * 需要被追踪并触发 fallback 的异常类型列表
     * 默认为 Throwable（即所有异常都会触发 fallback）
     */
    Class<? extends Throwable>[] exceptionsToTrace() default {Throwable.class};

    /**
     * 指定需要忽略的异常类型（即这些异常不会触发 fallback）
     * 注意：exceptionsToTrace 和 exceptionsToIgnore 不应同时使用；
     * 若同时存在，exceptionsToIgnore 优先级更高
     */
    Class<? extends Throwable>[] exceptionsToIgnore() default {};
}

实际使用示例

下面是一个完整的使用示例，展示了 @SentinelResource 注解的各种配置方式：

@RestController
public class SentinelController {

    @Autowired
    private ISentinelService service;

    @GetMapping(value = "/hello/{s}")
    public String apiHello(@PathVariable long s) {
        return service.hello(s);
    }
}

public interface ISentinelService {
    String hello(long s);
}

@Service
@Slf4j
public class SentinelServiceImpl implements ISentinelService {

    /**
     * Sentinel 提供了 @SentinelResource 注解用于定义资源
     *
     * @param s 输入参数
     * @return 返回结果
     */
    @Override
    // value：资源名称，必需项（不能为空）
    // blockHandler：对应处理 BlockException 的函数名称
    // fallback：用于在抛出异常的时候提供 fallback 处理逻辑
    @SentinelResource(value = "hello", blockHandler = "exceptionHandler", fallback = "helloFallback")
    public String hello(long s) {
        log.error("hello:{}", s);
        return String.format("Hello at %d", s);
    }

    /**
     * Fallback 函数
     * 函数签名与原函数一致，或加一个 Throwable 类型的参数
     */
    public String helloFallback(long s) {
        log.error("helloFallback:{}", s);
        return String.format("Halooooo %d", s);
    }

    /**
     * Block 异常处理函数
     * 参数最后多一个 BlockException，其余与原函数一致
     */
    public String exceptionHandler(long s, BlockException ex) {
        // Do some log here.
        log.error("exceptionHandler:{}", s);
        ex.printStackTrace();
        return "Oops, error occurred at " + s;
    }
}

SentinelResourceAspect 核心逻辑

@SentinelResource 注解由 SentinelResourceAspect 切面处理，核心逻辑如下：

@Aspect
public class SentinelResourceAspect extends AbstractSentinelAspectSupport {

    @Pointcut("@annotation(com.alibaba.csp.sentinel.annotation.SentinelResource)")
    public void sentinelResourceAnnotationPointcut() {
    }

    @Around("sentinelResourceAnnotationPointcut()")
    public Object invokeResourceWithSentinel(ProceedingJoinPoint pjp) throws Throwable {
        // 获取目标方法
        Method originMethod = resolveMethod(pjp);

        // 获取注解信息
        SentinelResource annotation = originMethod.getAnnotation(SentinelResource.class);
        if (annotation == null) {
            throw new IllegalStateException("Wrong state for SentinelResource annotation");
        }

        // 获取资源配置信息
        String resourceName = getResourceName(annotation.value(), originMethod);
        EntryType entryType = annotation.entryType();
        int resourceType = annotation.resourceType();

        Entry entry = null;
        try {
            // 创建限流入口
            entry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs());
            // 执行原方法
            Object result = pjp.proceed();
            return result;

        } catch (BlockException ex) {
            // 处理被限流异常
            return handleBlockException(pjp, annotation, ex);

        } catch (Throwable ex) {
            // 处理业务异常
            Class<? extends Throwable>[] exceptionsToIgnore = annotation.exceptionsToIgnore();
            // 优先检查忽略列表
            if (exceptionsToIgnore.length > 0 && exceptionBelongsTo(ex, exceptionsToIgnore)) {
                throw ex;
            }
            // 检查异常是否在追踪列表中
            if (exceptionBelongsTo(ex, annotation.exceptionsToTrace())) {
                traceException(ex);
                // 执行 fallback 逻辑
                return handleFallback(pjp, annotation, ex);
            }

            // 没有 fallback 函数可以处理该异常，直接抛出
            throw ex;

        } finally {
            // 释放 Entry 资源
            if (entry != null) {
                entry.exit(1, pjp.getArgs());
            }
        }
    }

    /**
     * 处理 BlockException
     *
     * blockHandler / blockHandlerClass 说明：
     * - blockHandler：对应处理 BlockException 的函数名称，可选项
     * - blockHandler 函数签名：与原方法相匹配并且最后加一个额外的参数，类型为 BlockException
     * - blockHandler 函数默认需要和原方法在同一个类中
     * - 若希望使用其他类的函数，则可以指定 blockHandlerClass 为对应的类的 Class 对象
     * - 注意：blockHandlerClass 中对应的函数必须为 static 函数，否则无法解析
     */
    protected Object handleBlockException(ProceedingJoinPoint pjp, SentinelResource annotation, BlockException ex)
            throws Throwable {

        // 执行 blockHandler 方法（如果配置了的话）
        Method blockHandlerMethod = extractBlockHandlerMethod(pjp, annotation.blockHandler(),
                annotation.blockHandlerClass());

        if (blockHandlerMethod != null) {
            Object[] originArgs = pjp.getArgs();
            // 构造参数：原方法参数 + BlockException
            Object[] args = Arrays.copyOf(originArgs, originArgs.length + 1);
            args[args.length - 1] = ex;

            try {
                // 根据 static 方法与否进行不同的调用
                if (isStatic(blockHandlerMethod)) {
                    return blockHandlerMethod.invoke(null, args);
                }
                return blockHandlerMethod.invoke(pjp.getTarget(), args);
            } catch (InvocationTargetException e) {
                // 抛出实际的异常
                throw e.getTargetException();
            }
        }

        // 如果没有 blockHandler，则尝试执行 fallback
        return handleFallback(pjp, annotation, ex);
    }

    /**
     * 处理 Fallback 逻辑
     *
     * fallback / fallbackClass 说明：
     * - fallback：fallback 函数名称，可选项，用于在抛出异常的时候提供 fallback 处理逻辑
     * - fallback 函数可以针对所有类型的异常（除了 exceptionsToIgnore 里面排除掉的异常类型）进行处理
     *
     * fallback 函数签名和位置要求：
     * - 返回值类型必须与原函数返回值类型一致
     * - 方法参数列表需要和原函数一致，或者可以额外多一个 Throwable 类型的参数用于接收对应的异常
     * - fallback 函数默认需要和原方法在同一个类中
     * - 若希望使用其他类的函数，则可以指定 fallbackClass 为对应的类的 Class 对象
     * - 注意：fallbackClass 中对应的函数必须为 static 函数，否则无法解析
     */
    protected Object handleFallback(ProceedingJoinPoint pjp, String fallback, String defaultFallback,
                                    Class<?>[] fallbackClass, Throwable ex) throws Throwable {
        Object[] originArgs = pjp.getArgs();

        // 执行 fallback 函数（如果配置了的话）
        Method fallbackMethod = extractFallbackMethod(pjp, fallback, fallbackClass);

        if (fallbackMethod != null) {
            // 构造参数：根据 fallback 方法的参数数量决定是否添加异常参数
            int paramCount = fallbackMethod.getParameterTypes().length;
            Object[] args;
            if (paramCount == originArgs.length) {
                args = originArgs;
            } else {
                args = Arrays.copyOf(originArgs, originArgs.length + 1);
                args[args.length - 1] = ex;
            }

            try {
                // 根据 static 方法与否进行不同的调用
                if (isStatic(fallbackMethod)) {
                    return fallbackMethod.invoke(null, args);
                }
                return fallbackMethod.invoke(pjp.getTarget(), args);
            } catch (InvocationTargetException e) {
                // 抛出实际的异常
                throw e.getTargetException();
            }
        }

        // 如果没有 fallback，尝试使用 defaultFallback
        return handleDefaultFallback(pjp, defaultFallback, fallbackClass, ex);
    }
}

五、流控处理核心逻辑

从入口函数开始，我们深入到流控处理的核心逻辑。

入口函数调用链

public class SphU {

    /**
     * 创建限流入口
     *
     * @param name 资源名称
     * @param resourceType 资源类型
     * @param trafficType 流量类型（IN 或 OUT）
     * @param args 参数数组
     * @return Entry 对象
     * @throws BlockException 如果被限流则抛出此异常
     */
    public static Entry entry(String name, int resourceType, EntryType trafficType, Object[] args)
            throws BlockException {
        return Env.sph.entryWithType(name, resourceType, trafficType, 1, args);
    }

    public static Entry entry(String name, EntryType trafficType, int batchCount) throws BlockException {
        return Env.sph.entry(name, trafficType, batchCount, OBJECTS0);
    }
}

public class CtSph implements Sph {

    @Override
    public Entry entry(String name, EntryType type, int count, Object... args) throws BlockException {
        StringResourceWrapper resource = new StringResourceWrapper(name, type);
        return entry(resource, count, args);
    }

    public Entry entry(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {
        return entryWithPriority(resourceWrapper, count, false, args);
    }

    /**
     * 带优先级的入口方法，这是限流的核心逻辑
     */
    private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
            throws BlockException {
        Context context = ContextUtil.getContext();

        // 如果上下文数量超过阈值，则不进行规则检查
        if (context instanceof NullContext) {
            // NullContext 表示上下文数量超过了阈值，这里只初始化 Entry，不进行规则检查
            return new CtEntry(resourceWrapper, null, context);
        }

        // 如果没有上下文，使用默认上下文
        if (context == null) {
            context = InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME);
        }

        // 如果全局开关关闭，则不进行规则检查
        if (!Constants.ON) {
            return new CtEntry(resourceWrapper, null, context);
        }

        // 获取或创建 ProcessorSlotChain（责任链）
        ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);

        /*
         * 如果资源（slot chain）数量超过 {@link Constants.MAX_SLOT_CHAIN_SIZE}，
         * 则不进行规则检查
         */
        if (chain == null) {
            return new CtEntry(resourceWrapper, null, context);
        }

        // 创建 Entry 对象
        Entry e = new CtEntry(resourceWrapper, chain, context);

        try {
            // 执行责任链进行规则检查
            chain.entry(context, resourceWrapper, null, count, prioritized, args);
        } catch (BlockException e1) {
            // 如果被限流，释放 Entry 并抛出异常
            e.exit(count, args);
            throw e1;
        } catch (Throwable e1) {
            // 这不应该发生，除非 Sentinel 内部存在错误
            log.warn("Sentinel unexpected exception,{}", e1.getMessage());
        }
        return e;
    }
}

ProcessorSlotChain 功能插槽链

lookProcessChain 方法实际创建了 ProcessorSlotChain 功能插槽链。ProcessorSlotChain 采用责任链模式，将不同的功能（限流、降级、系统保护）组合在一起。

SlotChain 的获取与创建

ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {
    // 先从缓存中获取
    ProcessorSlotChain chain = chainMap.get(resourceWrapper);

    if (chain == null) {
        // 双重检查锁，保证线程安全
        synchronized (LOCK) {
            chain = chainMap.get(resourceWrapper);
            if (chain == null) {
                // Entry 大小限制
                if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) {
                    return null;
                }

                // 创建新的 SlotChain
                chain = SlotChainProvider.newSlotChain();

                // 使用不可变模式更新缓存
                Map<ResourceWrapper, ProcessorSlotChain> newMap =
                    new HashMap<ResourceWrapper, ProcessorSlotChain>(chainMap.size() + 1);
                newMap.putAll(chainMap);
                newMap.put(resourceWrapper, chain);
                chainMap = newMap;
            }
        }
    }
    return chain;
}

SlotChain 的构建

public class DefaultSlotChainBuilder implements SlotChainBuilder {

    @Override
    public ProcessorSlotChain build() {
        ProcessorSlotChain chain = new DefaultProcessorSlotChain();

        // 通过 SPI 加载所有 ProcessorSlot 并排序
        List<ProcessorSlot> sortedSlotList = SpiLoader.loadPrototypeInstanceListSorted(ProcessorSlot.class);

        for (ProcessorSlot slot : sortedSlotList) {
            // 只处理继承自 AbstractLinkedProcessorSlot 的 Slot
            if (!(slot instanceof AbstractLinkedProcessorSlot)) {
                RecordLog.warn("The ProcessorSlot(" + slot.getClass().getCanonicalName() +
                    ") is not an instance of AbstractLinkedProcessorSlot, can't be added into ProcessorSlotChain");
                continue;
            }

            // 将 Slot 添加到责任链尾部
            chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
        }

        return chain;
    }
}

SlotChain 的功能划分

Slot Chain 可以分为两部分：

• 统计数据构建部分（statistic）：负责收集各种指标数据；
• 判断部分（rule checking）：根据规则判断是否限流。

官方架构图很好地解释了各个 Slot 的作用及其负责的部分。目前 ProcessorSlotChain 的设计是一个资源对应一个，构建好后缓存起来，方便下次直接取用。

各 Slot 的执行顺序

以下是 Sentinel 中各个 Slot 的默认执行顺序：

NodeSelectorSlot
    ↓
ClusterBuilderSlot
    ↓
StatisticSlot
    ↓
ParamFlowSlot
    ↓
SystemSlot
    ↓
AuthoritySlot
    ↓
FlowSlot
    ↓
DegradeSlot

NodeSelectorSlot - 上下文节点选择

这个功能插槽主要为资源下不同的上下文创建对应的 DefaultNode（实际用于统计指标信息）。解释一下Sentinel中的Node是什么，简单来说就是每个资源统计指标存放的容器，只不过内部由于不同的统计口径（秒级、分钟及）而分别有不同的统计窗口。Node在Sentinel不是单一的结构，而是总体上形成父子关系的树形结构。

不同的调用会有不同的 context 名称，如在当前 MVC 场景下，上下文为 sentinel_web_servlet_context。

public class NodeSelectorSlot extends AbstractLinkedProcessorSlot<Object> {

    /**
     * 同一个资源在不同上下文中的 DefaultNode 映射
     */
    private volatile Map<String, DefaultNode> map = new HashMap<String, DefaultNode>(10);

    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, Object obj, int count,
                      boolean prioritized, Object... args) throws Throwable {
        // 从映射表中获取当前上下文对应的节点
        DefaultNode node = map.get(context.getName());

        if (node == null) {
            // 双重检查锁，保证线程安全
            synchronized (this) {
                node = map.get(context.getName());
                if (node == null) {
                    // 创建新的 DefaultNode
                    node = new DefaultNode(resourceWrapper, null);

                    // 使用写时复制更新缓存
                    HashMap<String, DefaultNode> cacheMap = new HashMap<String, DefaultNode>(map.size());
                    cacheMap.putAll(map);
                    cacheMap.put(context.getName(), node);
                    map = cacheMap;

                    // 构建调用树
                    ((DefaultNode) context.getLastNode()).addChild(node);
                }
            }
        }

        // 设置当前上下文的当前节点
        context.setCurNode(node);
        // 继续执行后续 Slot
        fireEntry(context, resourceWrapper, node, count, prioritized, args);
    }

    @Override
    public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
        fireExit(context, resourceWrapper, count, args);
    }
}

ClusterBuilderSlot - 集群节点构建

这个功能槽主要用于创建 ClusterNode。ClusterNode 和 DefaultNode 的区别是：

DefaultNode 是特定于上下文的（context-specific）；

ClusterNode 是不区分上下文的（context-independent），用于统计该资源在所有上下文中的整体数据。

public class ClusterBuilderSlot extends AbstractLinkedProcessorSlot<DefaultNode> {

    /**
     * 全局 ClusterNode 映射表
     */
    private static volatile Map<ResourceWrapper, ClusterNode> clusterNodeMap = new HashMap<>();

    private static final Object lock = new Object();

    private volatile ClusterNode clusterNode = null;

    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                      boolean prioritized, Object... args) throws Throwable {
        // 创建 ClusterNode（如果不存在）
        if (clusterNode == null) {
            synchronized (lock) {
                if (clusterNode == null) {
                    // 创建集群节点
                    clusterNode = new ClusterNode(resourceWrapper.getName(), resourceWrapper.getResourceType());

                    // 更新全局映射表
                    HashMap<ResourceWrapper, ClusterNode> newMap =
                        new HashMap<>(Math.max(clusterNodeMap.size(), 16));
                    newMap.putAll(clusterNodeMap);
                    newMap.put(node.getId(), clusterNode);

                    clusterNodeMap = newMap;
                }
            }
        }

        // 将 ClusterNode 设置到 DefaultNode 中
        node.setClusterNode(clusterNode);

        // 如果有来源标识，则创建 origin node
        if (!"".equals(context.getOrigin())) {
            Node originNode = node.getClusterNode().getOrCreateOriginNode(context.getOrigin());
            context.getCurEntry().setOriginNode(originNode);
        }

        // 继续执行后续 Slot
        fireEntry(context, resourceWrapper, node, count, prioritized, args);
    }
}

StatisticSlot - 统计插槽

StatisticSlot 是 Sentinel 最重要的类之一，用于根据规则判断结果进行相应的统计操作。

统计逻辑说明

entry 的时候：

依次执行后续的判断 Slot；

每个 Slot 触发流控会抛出异常（BlockException 的子类）；

若有 BlockException 抛出，则记录 block 数据；

若无异常抛出则算作可通过（pass），记录 pass 数据。

exit 的时候：

若无 error（无论是业务异常还是流控异常），记录 complete（success）以及 RT，线程数 -1。

记录数据的维度：

线程数 +1；

记录当前 DefaultNode 数据；

记录对应的 originNode 数据（若存在 origin）；

累计 IN 统计数据（若流量类型为 IN）。

public class StatisticSlot extends AbstractLinkedProcessorSlot<DefaultNode> {

    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                      boolean prioritized, Object... args) throws Throwable {
        try {
            // 此位置会调用 SlotChain 中后续的所有 Slot，完成所有规则检测
            fireEntry(context, resourceWrapper, node, count, prioritized, args);

            // 请求通过，增加线程数和通过数
            // 代码运行到这个位置，就证明之前的所有 Slot 检测都通过了
            // 此时就可以统计请求的相应数据了

            // 增加线程数（+1）
            node.increaseThreadNum();
            // 增加通过请求的数量（这里涉及到滑动窗口算法）
            node.addPassRequest(count);

            // 省略其他统计逻辑...

        } catch (PriorityWaitException ex) {
            // 如果是优先级等待异常，记录优先级等待数
            node.increaseThreadNum();
            if (context.getCurEntry().getOriginNode() != null) {
                context.getCurEntry().getOriginNode().increaseThreadNum();
            }
            if (resourceWrapper.getEntryType() == EntryType.IN) {
                // 记录入站统计数据
                Constants.ENTRY_NODE.increaseThreadNum();
            }
            throw ex;

        } catch (BlockException e) {
            // 如果被限流，记录被限流数
            // 省略 block 统计逻辑...
            throw e;

        } catch (Throwable ex) {
            // 如果发生业务异常，记录异常数
            // 省略异常统计逻辑...
            throw ex;
        }
    }

    @Override
    public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
        // 若无 error（无论是业务异常还是流控异常），记录 complete（success）以及 RT，线程数-1
        // 记录数据的维度：线程数+1、记录当前 DefaultNode 数据、记录对应的 originNode 数据（若存在 origin）
        // 、累计 IN 统计数据（若流量类型为 IN）
        // 省略 exit 统计逻辑...
    }
}

StatisticNode 数据结构

到这里，StatisticSlot 的作用已经比较清晰了。接下来我们需要分析它的统计数据结构。fireEntry 调用向下的节点和之前的方式一样，剩下的节点主要包括：

• ParamFlowSlot；
• SystemSlot；
• AuthoritySlot；
• FlowSlot；
• DegradeSlot；

其中比较常见的是流控和熔断：FlowSlot、DegradeSlot，所以下面我们着重分析 FlowSlot。

六、FlowSlot - 流控插槽

这个 Slot 主要根据预设的资源的统计信息，按照固定的次序依次生效。如果一个资源对应两条或者多条流控规则，则会根据如下次序依次检验，直到全部通过或者有一个规则生效为止。

FlowSlot 核心逻辑

@SpiOrder(-2000)
public class FlowSlot extends AbstractLinkedProcessorSlot<DefaultNode> {

    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                      boolean prioritized, Object... args) throws Throwable {
        // 执行流控检查
        checkFlow(resourceWrapper, context, node, count, prioritized);

        // 继续执行后续 Slot
        fireEntry(context, resourceWrapper, node, count, prioritized, args);
    }

    // 省略其他方法...
}

checkFlow 方法详解

/**
 * 执行流控检查
 *
 * @param ruleProvider 规则提供者函数
 * @param resource 资源包装器
 * @param context 上下文
 * @param node 节点
 * @param count 请求数量
 * @param prioritized 是否优先
 * @throws BlockException 如果被限流则抛出异常
 */
public void checkFlow(Function<String, Collection<FlowRule>> ruleProvider, ResourceWrapper resource,
                      Context context, DefaultNode node, int count, boolean prioritized) throws BlockException {
    // 判断规则和资源不能为空
    if (ruleProvider == null || resource == null) {
        return;
    }

    // 获取指定资源的所有流控规则
    Collection<FlowRule> rules = ruleProvider.apply(resource.getName());

    // 逐个应用流控规则。若无法通过则抛出异常，后续规则不再应用
    if (rules != null) {
        for (FlowRule rule : rules) {
            if (!canPassCheck(rule, context, node, count, prioritized)) {
                // FlowException 继承 BlockException
                throw new FlowException(rule.getLimitApp(), rule);
            }
        }
    }
}

通过这里我们就可以得知，流控规则是通过 FlowRule 来完成的，数据来源是我们使用的流控控制台，也可以通过代码进行设置。

FlowRule 流控规则

每条流控规则主要由三个要素构成：

• grade（阈值类型）：按 QPS（每秒请求数）还是线程数进行限流;
• strategy（调用关系策略）：基于调用关系的流控策略;
• controlBehavior（流控效果）：当 QPS 超过阈值时的流量整形行为。

public class FlowRule extends AbstractRule {

    public FlowRule() {
        super();
        // 来源默认 Default
        setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);
    }

    public FlowRule(String resourceName) {
        super();
        // 资源名称
        setResource(resourceName);
        setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);
    }

    /**
     * 流控的阈值类型
     * 0: 线程数
     * 1: QPS
     */
    private int grade = RuleConstant.FLOW_GRADE_QPS;

    /**
     * 流控阈值
     */
    private double count;

    /**
     * 基于调用链的流控策略
     * STRATEGY_DIRECT: 直接流控（按来源）
     * STRATEGY_RELATE: 关联流控（关联资源）
     * STRATEGY_CHAIN: 链路流控（按入口资源）
     */
    private int strategy = RuleConstant.STRATEGY_DIRECT;

    /**
     * 关联流控模式下的关联资源
     */
    private String refResource;

    /**
     * 流控效果（流量整形行为）
     * 0: 默认（直接拒绝）
     * 1: 预热（Warm Up）
     * 2: 排队等待（Rate Limiter）
     * 3: 预热 + 排队等待（目前控制台没有）
     */
    private int controlBehavior = RuleConstant.CONTROL_BEHAVIOR_DEFAULT;

    /**
     * 预热时长（秒）
     */
    private int warmUpPeriodSec = 10;

    /**
     * 排队等待的最大超时时间（毫秒）
     */
    private int maxQueueingTimeMs = 500;

    /**
     * 是否为集群模式
     */
    private boolean clusterMode;

    /**
     * 集群模式配置
     */
    private ClusterFlowConfig clusterConfig;

    /**
     * 流量整形控制器
     */
    private TrafficShapingController controller;

    // 省略 getter/setter 方法...
}

七、滑动窗口算法

不管流控规则采用何种流控算法，在底层都需要有支持指标统计的数据结构作为支撑。在 Sentinel 中，用于支撑基于 QPS 等限流的数据结构是 StatisticNode。

StatisticNode 数据结构

public class StatisticNode implements Node {

    /**
     * 保存最近 1 秒内的统计数据
     * 每个桶（bucket）500ms，共 2 个桶
     */
    private transient volatile Metric rollingCounterInSecond =
        new ArrayMetric(SampleCountProperty.SAMPLE_COUNT, IntervalProperty.INTERVAL);

    /**
     * 保存最近 60 秒的统计数据
     * windowLengthInMs 被特意设置为 1000 毫秒，即每个桶代表 1 秒
     * 共 60 个桶，这样可以获得每秒精确的统计信息
     */
    private transient Metric rollingCounterInMinute =
        new ArrayMetric(60, 60 * 1000, false);

    // 省略其他字段和方法...
}

ArrayMetric 核心实现

ArrayMetric 是 Sentinel 中数据采集的核心，内部使用了 BucketLeapArray，即滑动窗口的思想进行数据的采集。

public class ArrayMetric implements Metric {

    /**
     * 滑动窗口数组
     */
    private final LeapArray<MetricBucket> data;

    public ArrayMetric(int sampleCount, int intervalInMs) {
        this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
    }

    public ArrayMetric(int sampleCount, int intervalInMs, boolean enableOccupy) {
        if (enableOccupy) {
            // 可抢占的滑动窗口，支持借用未来窗口的配额
            this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
        } else {
            // 普通滑动窗口
            this.data = new BucketLeapArray(sampleCount, intervalInMs);
        }
    }
}

这里有两种实现：

• BucketLeapArray：普通滑动窗口，每个时间桶仅记录固定时间窗口内的指标数据；
• OccupiableBucketLeapArray：扩展实现，支持"抢占"未来时间窗口的令牌或容量，在流量突发时允许借用后续窗口的配额，实现更平滑的限流效果。

BucketLeapArray - 滑动窗口实现

LeapArray 核心属性

LeapArray 是滑动窗口的基础类，其核心属性如下：

/**
 * 窗口大小（长度），单位：毫秒
 * 例如：1000ms
 */
private int windowLengthInMs;

/**
 * 样本数（桶的数量）
 * 例如：5（表示 5 个桶，每个 1000ms，总共 5 秒）
 */
private int sampleCount;

/**
 * 采集周期（总时间窗口长度），单位：毫秒
 * 例如：5 * 1000ms（5 秒）
 */
private int intervalInMs;

/**
 * 窗口数组，array 长度就是样本数 sampleCount
 */
protected final AtomicReferenceArray<WindowWrap<T>> array;

/**
 * 更新窗口数据的锁，保证数据的正确性
 */
private final ReentrantLock updateLock;

WindowWrap 窗口包装器

每个窗口包装器包含三个属性：

 public class WindowWrap<T> {

    /**
     * 窗口大小（长度），单位：毫秒
     * 与 LeapArray 中的 windowLengthInMs 一致
     */
    private final long windowLengthInMs;

    /**
     * 窗口开始时间戳
     * 它的值是 windowLengthInMs 的整数倍
     */
    private long windowStart;

    /**
     * 窗口数据（泛型 T）
     * Sentinel 目前只有 MetricBucket 类型，存储统计数据
     */
    private T value;
}

MetricBucket 指标桶

public class MetricBucket {

    /**
     * 计数器数组
     * 长度是需要统计的事件种类数，目前是 6 个
     * LongAdder 是线程安全的计数器，性能优于 AtomicLong
     */
    private final LongAdder[] counters;
    
    // 省略其他字段和方法...
}

滑动窗口工作原理

LeapArray 统计数据的基本思路：

创建一个长度为 n 的数组，数组元素就是窗口；

每个窗口包装了 1 个指标桶，桶中存放了该窗口时间范围内对应的请求统计数据；

可以想象成一个环形数组在时间轴上向右滚动；

请求到达时，会命中数组中的一个窗口，该请求的数据就会存到命中的这个窗口包含的指标桶中；

当数组转满一圈时，会回到数组的开头；

此时下标为 0 的元素需要重复使用，它里面的窗口数据过期了，需要重置，然后再使用。

获取当前窗口

LeapArray 获取当前时间窗口的方法：

 /**
 * 获取当前时间戳对应的窗口
 *
 * @return 当前时间的窗口
 */
public WindowWrap<T> currentWindow() {
    return currentWindow(TimeUtil.currentTimeMillis());
}

/**
 * 获取指定时间戳对应的窗口（核心方法）
 *
 * @param timeMillis 时间戳（毫秒）
 * @return 对应的窗口
 */
public WindowWrap<T> currentWindow(long timeMillis) {
    if (timeMillis < 0) {
        return null;
    }

    // 计算数组下标
    int idx = calculateTimeIdx(timeMillis);

    // 计算当前请求对应的窗口开始时间
    long windowStart = calculateWindowStart(timeMillis);

    // 无限循环，确保能够获取到窗口
    while (true) {
        // 取窗口
        WindowWrap<T> old = array.get(idx);

        if (old == null) {
            // 第一次使用，创建新窗口
            WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));

            // CAS 操作，确保只初始化一次
            if (array.compareAndSet(idx, null, window)) {
                // 成功更新，返回创建的窗口
                return window;
            } else {
                // CAS 失败，让出时间片，等待其他线程完成初始化
                Thread.yield();
            }

        } else if (windowStart == old.windowStart()) {
            // 命中：取出的窗口的开始时间和本次请求计算出的窗口开始时间一致
            return old;

        } else if (windowStart > old.windowStart()) {
            // 窗口过期：本次请求计算出的窗口开始时间大于取出的窗口
            // 说明取出的窗口过期了，需要重置
            if (updateLock.tryLock()) {
                try {
                    // 成功获取锁，更新窗口开始时间，计数器重置
                    return resetWindowTo(old, windowStart);
                } finally {
                    updateLock.unlock();
                }
            } else {
                // 获取锁失败，让出时间片，等待其他线程更新
                Thread.yield();
            }

        } else if (windowStart < old.windowStart()) {
            // 异常情况：机器时钟回拨等
            // 正常情况不会进入该分支
            return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
        }
    }
}

数据存储

在获取到窗口之后，就可以存储数据了。ArrayMetric 实现了 Metric 中存取数据的接口方法。

示例：存储 RT（响应时间）

/**
 * 添加响应时间数据
 *
 * @param rt 响应时间（毫秒）
 */
public void addRT(long rt) {
    // 获取当前时间窗口，data 为 BucketLeapArray
    WindowWrap<MetricBucket> wrap = data.currentWindow();

    // 计数
    wrap.value().addRT(rt);
}

/**
 * MetricBucket 的 addRT 方法
 *
 * @param rt 响应时间
 */
public void addRT(long rt) {
    // 记录 RT 时间对 rt 值
    add(MetricEvent.RT, rt);

    // 记录最小响应时间（非线程安全，但没关系）
    if (rt < minRt) {
        minRt = rt;
    }
}

/**
 * 通用的计数方法
 *
 * @param event 事件类型
 * @param n 增加的数量
 * @return 当前桶
 */
public MetricBucket add(MetricEvent event, long n) {
    counters[event.ordinal()].add(n);
    return this;
}

数据读取

示例：读取 RT（响应时间）

/**
 * 获取总响应时间
 *
 * @return 总响应时间
 */
public long rt() {
    // 触发当前窗口更新（处理过期窗口）
    data.currentWindow();

    long rt = 0;
    // 取出所有的 bucket
    List<MetricBucket> list = data.values();

    for (MetricBucket window : list) {
        rt += window.rt(); // 求和
    }
    return rt;
}

/**
 * 获取所有有效的窗口
 *
 * @return 有效窗口列表
 */
public List<T> values() {
    return values(TimeUtil.currentTimeMillis());
}

/**
 * 获取指定时间之前的所有有效窗口
 *
 * @param timeMillis 时间戳
 * @return 有效窗口列表
 */
public List<T> values(long timeMillis) {
    if (timeMillis < 0) {
        return new ArrayList<T>(); // 正常情况不会到这里
    }

    int size = array.length();
    List<T> result = new ArrayList<T>(size);

    for (int i = 0; i < size; i++) {
        WindowWrap<T> windowWrap = array.get(i);

        // 过滤掉没有初始化过的窗口和过期的窗口
        if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) {
            continue;
        }

        result.add(windowWrap.value());
    }
    return result;
}

/**
 * 判断窗口是否过期
 *
 * @param time 给定时间（通常是当前时间）
 * @param windowWrap 窗口包装器
 * @return 如果过期返回 true
 */
public boolean isWindowDeprecated(long time, WindowWrap<T> windowWrap) {
    // 给定时间与窗口开始时间超过了一个采集周期
    return time - windowWrap.windowStart() > intervalInMs;
}

OccupiableBucketLeapArray - 可抢占窗口

为什么需要 OccupiableBucketLeapArray？

假设一个资源的访问 QPS 稳定是 10，请求是均匀分布的：

在时间 0.0-1.0 秒区间中，通过了 10 个请求；

在 1.1 秒的时候，观察到的 QPS 可能只有 5，因为此时第一个时间窗口被重置了，只有第二个时间窗口有值；

当在秒级统计的情形下，用 BucketLeapArray 会有 0~50%的数据误这时就要用 OccupiableBucketLeapArray 来解决这个问题。

OccupiableBucketLeapArray 实现

从上面我们可以看到在秒级统计 rollingCounterInSecond 中，初始化实例时有两种构造参数：

public class OccupiableBucketLeapArray extends LeapArray<MetricBucket> {

    /**
     * 借用未来窗口的数组
     */
    private final FutureBucketLeapArray borrowArray;

    public OccupiableBucketLeapArray(int sampleCount, int intervalInMs) {
        super(sampleCount, intervalInMs);
        // 创建借用窗口数组
        this.borrowArray = new FutureBucketLeapArray(sampleCount, intervalInMs);
    }

    /**
     * 创建新的空桶
     * 会从 borrowArray 中借用数据
     */
    @Override
    public MetricBucket newEmptyBucket(long time) {
        MetricBucket newBucket = new MetricBucket();

        // 获取借用窗口的数据
        MetricBucket borrowBucket = borrowArray.getWindowValue(time);
        if (borrowBucket != null) {
            // 将借用数据复制到新桶中
            newBucket.reset(borrowBucket);
        }

        return newBucket;
    }

    /**
     * 重置窗口
     * 会从 borrowArray 中借用 pass 数据
     */
    @Override
    protected WindowWrap<MetricBucket> resetWindowTo(WindowWrap<MetricBucket> w, long time) {
        // 更新开始时间并重置值
        w.resetTo(time);

        MetricBucket borrowBucket = borrowArray.getWindowValue(time);
        if (borrowBucket != null) {
            // 重置桶值并添加借用的 pass 数据
            w.value().reset();
            w.value().addPass((int) borrowBucket.pass());
        } else {
            w.value().reset();
        }

        return w;
    }

    /**
     * 获取当前等待中的请求数量
     */
    @Override
    public long currentWaiting() {
        borrowArray.currentWindow();
        long currentWaiting = 0;
        List<MetricBucket> list = borrowArray.values();

        for (MetricBucket window : list) {
            currentWaiting += window.pass();
        }
        return currentWaiting;
    }

    /**
     * 添加等待中的请求数量
     *
     * @param time 时间
     * @param acquireCount 获取数量
     */
    @Override
    public void addWaiting(long time, int acquireCount) {
        WindowWrap<MetricBucket> window = borrowArray.currentWindow(time);
        window.value().add(MetricEvent.PASS, acquireCount);
    }
}

八、总结

至此，Sentinel 的基本情况都已经分析完成。以上内容主要讲解了 Sentinel 的核心处理流程，包括：

核心流程总结

1. 规则加载：

• 通过 RuleLoader 将各种规则（流控、熔断、系统限流等）加载到内存缓存中。

2. 请求拦截：

• 通过 SentinelServletFilter 过滤器拦截 HTTP 请求；
• 通过SentinelResourceAspect切面处理 @SentinelResource 注解。

3. 责任链处理：

• 使用 ProcessorSlotChain 责任链模式组合多个功能插槽；
• 每个插槽负责特定的功能（统计、流控、熔断等）。

4. 流控判断：

• FlowSlot 根据流控规则判断是否限流；
• 通过滑动窗口算法统计 QPS、线程数等指标。

5. 异常处理：

• 被限流时抛出 BlockException；
• 通过 blockHandler 或 fallback 处理异常。

核心技术点

1. 责任链模式：

• 通过 ProcessorSlotChain 将不同的限流功能组合在一起。

2. 滑动窗口算法：

• LeapArray 实现环形滑动窗口；
• BucketLeapArray 普通滑动窗口；
• OccupiableBucketLeapArray 可抢占窗口，支持借用未来配额。

3. 数据结构：

• DefaultNode：特定于上下文的统计节点；
• ClusterNode：不区分上下文的集群统计节点；
• StatisticNode：核心统计节点，包含秒级和分钟级统计。

4. 限流算法：

• QPS 限流：通过滑动窗口统计 QPS；
• 线程数限流：通过原子计数器统计线程数；
• 流控效果：快速失败、预热、排队等待等；

Sentinel 通过精心设计的架构，实现了高效、灵活、可扩展的流量控制能力，为微服务系统提供了强大的保护机制。

往期回顾

1.社区推荐重排技术：双阶段框架的实践与演进｜得物技术

2.Flink ClickHouse Sink：生产级高可用写入方案｜得物技术

3.服务拆分之旅：测试过程全揭秘｜得物技术

4.大模型网关：大模型时代的智能交通枢纽｜得物技术

5.从“人治”到“机治”：得物离线数仓发布流水线质量门禁实践

文 /万钧

关注得物技术，每周更新技术干货

要是觉得文章对你有帮助的话，欢迎评论转发点赞～

未经得物技术许可严禁转载，否则依法追究法律责任。

一、背景

推荐系统典型pipeline

在推荐系统多阶段Pipeline（召回→粗排→精排→重排）中，重排作为最终决策环节，承担着将精排输出的有限候选集（通常为Top 100–500个Item）转化为最优序列的关键职责。数学定义为在给定候选集$C = \lbrace x_1,x_2,……,x_n \rbrace$C={x1,x2,……,xn}与目标列表长度$L$L，重排的目标是寻找一个排列$\pi^* \in P(C,L)$π∗∈P(C,L)，使得全局收益函数最大化。

在推荐系统、搜索排序等AI领域，Pointwise 建模是精排阶段的核心方法，即对每个 Item 独立打分后排序，pointwise 建模范式面临挑战：

• 多样性约束：精排按 item 独立打分排序 → 高分 item 往往语义/类目高度同质（如5个相似短视频连续曝光）。
• 位置偏差：用户注意力随位置显著衰减，且不同item对位置敏感度不同。
• 上下文建模：用户决策是序列行为，而非独立事件。

二、重排架构演进：生成式模型实践

我们的重排系统采用G-E两阶段协同框架：

• 生成阶段（Generation）：高效生成若干高质量候选排列。
• 评估阶段（Evaluation）：对候选排列进行精细化打分，选出全局最优结果。

不考虑算力和耗时的情况下，通过穷举所有排列$P(C,L)$P(C,L)。

生成阶段主要依赖启发式规则、随机扰动 + beamSearch算法生成候选list，双阶段范式存在显著的痛点：

• 质量-延迟-多样性的“不可能三角”：在实践中，增加生成候选list数一般可以提升最终list的质量，但边际收益递减；优化过程中，我们通过增加多目标、多样性等策略都取得了消费指标的提升，但在候选list达到百量级时，单纯增加候选集对指标的提升，同时还有：

1. 增加beam width，系统耗时增加，DCG@K提升逐渐减少。

2. 增加通道数，通道间重叠度逐渐增加，去重list增加逐渐减少。

• 阶段间目标不一致：

1. 分布偏移：启发式生成Beam Search输出的Top排列中，20%被评估模型否定，生成阶段搜索效率浪费。

2. 梯度断层：Beam Search含argmax操作，双阶段无法端到端优化；生成模型无法感知评估反馈，优化方向偏离全局最优。

生成模型优化

生成分为启发式方法和生成式模型方法, 一般认为生成式模型方法要好于启发式方法。生成式模型逐渐成为重排主流范式，主要分为两类：自回归生成模型、非自回归生成模型。

• 自回归生成：按位置顺序逐个生成物品，第 t 位的预测依赖前 t-1 位已生成结果。

1. 优点：

a. 序列依赖建模强，天然捕获物品间的顺序依赖。

b. 训练简单稳定，每步使用真实前序作为输入，收敛快。

c. 生成质量高，逐步细化决策，适合长序列精细优化。

2. 缺点：

a. 推理延迟高，生成 L 个物品需 L 次前向传播，线上服务难以满足毫秒级要求。

b. 局部最优风险，早期错误决策无法回溯修正，影响整体序列质量。

• 非自回归生成：一次性预测整个推荐序列的所有位置，各位置预测相互独立。

1. 优点：

a. 推理速度极快：生成整个序列仅需1次前向传播。

2.缺点：

b.条件独立性假设过强：各位置并行预测，难以显式建模物品间复杂依赖关系。

非自回归模型

为了对齐双阶段一致性，同时考虑线上性能，我们推进了非自回归模型的上线。模型结构如下图：

模型包括Candidates Encoder和Position Encoder，Candidates Encoder是标准的Transformer结构, 用于获取item间的交互信息；Position Encoder额外增加了Cross Attention，期望Position序列同时关注Candidate序列。

• 模型特征：用户信息、item特征、位置信息、上游精排打分特征。
• 模型输出：一次性输出 n×L 的位置-物品得分矩阵（n 为候选 item 数，L 为目标列表长度），支持高效并行推理

• 位置感知建模：引入可学习位置嵌入，显式建模“同一 item 在不同位置表现不同”的现象（如首屏效应、位置衰减）。
• 训练目标：模型使用logloss，让正反馈label序列的生成概率最大, 同时负反馈label序列的生成概率最小：

其中，$p_{ij}$pij表示位置i上是否展示物品j，$y_{i}$yi表示位置i上的label。

线上实验及收益：

• 一期新增了非自回归生成通道，pvctr +0.6%，时长+0.55%。
• 二期在所有通道排序因子中bagging非自回归模型，pvctr +1.0%，时长+1.13%。

自回归模型

由于条件独立性假设, 非自回归模型对上下文信息建模是不够的，近期我们重点推进了自回归模型的开发。

模型通过Transformer架构建模list整体收益，我们使用单向transformer模拟用户浏览行为的因果性，同时解决自回归生成的暴露偏差问题，保持训练和推理的一致性。结构如下：

• 模型特征：用户信息、item特征、位置信息、上游精排打分特征。
• 训练目标：模型使用有序回归loss，在评估多个回合中不同长度的子列表时，能够很好地体现出序列中的增量价值。是用于判断长度为j的子列表是否已经达到i次点击或转化的损失函数。

线上模型推理效率优化及实验效果：

自回归生成模型推理延迟高，生成 L 个物品需 L 次前向传播，线上服务难以满足毫秒级要求。因此，我们在传统自回归生成模型的基础上增加MTP（multi token prediction）结构，突破生成式重排模型推理瓶颈。其核心思想是将传统自回归的单步预测扩展为单步多token联合预测，显著减少生成迭代次数。

自回归生成模型在社区推荐已完成了推全，实验中我们新增了自回归生成模型通道，但不是完全体，仅部分位置生成调用了模型：

• 一期调用两次模型，每次预测4个位置，pvctr +0.69%，有效vv +0.58%。
• 二期调用两次模型，每次预测5个位置，pvctr +0.54%，有效vv +0.40%。

三、推理性能优化：端到端生成的效率保障

工程架构

为解决CPU推理模型延迟高、制约业务效果的问题，我们对DScatter模型服务进行升级，引入高性能GPU推理能力，具体方案如下：

• GPU推理框架集成与升级：

1. 框架升级：将现有依赖的推理框架升级为支持GPU的高性能服务框架。

2. 硬件资源引入：引入 NVIDIA L20 等专业推理显卡，为当前的listwise评估模型及自回归生成模型提供专用算力，实现模型推理的硬件加速。

• DScatter模型服务独立部署与容量提升：

1. 为解决模型部署效率低与资源竞争问题，将DScatter的模型打分逻辑从现有重排服务中完全解耦，构建并部署独立的 DScatter-Model-Server 集群，从根本上消除与重排服务在CPU、内存等关键资源上的竞争。

模型优化

• 模型格式转换与加速：

导出为 ONNX 格式，使用 TensorRT 进行量化、层融合、动态张量显存等技术加速推理。

• Item Embedding缓存：

预计算item静态网络，线上直接查询节省计算量。

• 自回归生成模型核心优化，KV Cache 复用：

缓存已生成token的KV和attention值，仅计算增量token相关值，避免重复计算。

• 其他LLM推理加速技术应用落地，例如GQA

四、未来规划：迈向端到端序列生成的下一代重排架构

当前“生成-评估”双阶段范式虽在工程落地性上取得平衡，但其本质仍是局部优化：生成阶段依赖启发式规则或浅层模型生成候选，评估阶段虽能识别优质序列，却无法反向指导生成过程，导致系统能力存在理论上限。为突破这一瓶颈，我们规划构建端到端序列生成（End-to-End Sequence Generation） 架构，将重排从“候选筛选”升级为“序列创造”，直接以全局业务目标（如用户停留时长、互动深度、内容生态健康度）为优化目标。

核心架构设计：

• 统一生成器：以 Transformer 为基础架构，搭建自回归序列建模能力，采用分层混合生成策略：

1. 粗粒度并行生成：首层预测序列骨架（如类目分布、内容密度）等。

2. 细粒度自回归精调：在骨架约束下，自回归生成具体 item，确保局部最优。

• 序列级Reward Modeling：

1. 构建多目标 reward 函数：xtr、多样性。

2. Engagement：基于用户滑动轨迹建模序列累积收益（如滑动深度加权CTR）。

3. Diversity：跨类目/创作者/内容形式的分布熵。

4.Fairness：冷启内容、长尾创作者曝光保障。

训练范式升级：强化学习与对比学习融合

推进自回归生成模型的架构升级与训练体系重构，引入强化学习微调（PPO/DPO）与对比学习机制，提升序列整体效率。

• 搭建近线系统，生成高质量list候选，提升系统能力上限：

1.基于 DCG 的列表质量打分：

a. 对每个曝光列表L，计算其 DCG@K作为质量分数：

其中 gain(item)可定义为：

若点击：+1.0

若互动（点赞/收藏）：+1.5

若观看 >5s：+0.8

否则：0

2.构造偏好对：

a.对同一用户在同一上下文下的两个列表 $L_w$Lw（win）和 $L_l$Ll（lose）。

b.若 $DCG(L_w) > DCG(L_l) + \delta$DCG(Lw)>DCG(Ll)+δ（δ 为 margin，如 0.1），则构成一个有效偏好对。

• 引入强化学习微调（PPO/DPO）与对比学习机制，提升序列整体效率：

1. 模型结构：

a.使用当前自回归生成模型作为策略模型。

b.固定预训练模型作为参考策略 （即 DPO 中的“旧策略”）。

2.DPO损失：

• 技术价值：

1. 突破“质量-延迟-多样性”不可能三角：通过序列级优化，在同等延迟下实现质量与多样性双提升。

2. 为AIGC与推荐融合铺路：端到端生成器可无缝接入AIGC内容，实现“内容生成-序列编排”一体化。

参考文献：

1. Gloeckle F, Idrissi B Y, Rozière B, et al. Better & faster large language models via multi-token prediction[J]. arXiv preprint arXiv:2404.19737, 2024.
2. Ren Y, Yang Q, Wu Y, et al. Non-autoregressive generative models for reranking recommendation[C]//Proceedings of the 30th ACM SIGKDD Conference on Knowledge Discovery and Data Mining. 2024: 5625-5634.
3. Meng Y, Guo C, Cao Y, et al. A generative re-ranking model for list-level multi-objective optimization at taobao[C]//Proceedings of the 48th International ACM SIGIR Conference on Research and Development in Information Retrieval. 2025: 4213-4218.
4. Zhao X, Xia L, Zhang L, et al. Deep reinforcement learning for page-wise recommendations[C]//Proceedings of the 12th ACM conference on recommender systems. 2018: 95-103.
5. Feng Y, Hu B, Gong Y, et al. GRN: Generative Rerank Network for Context-wise Recommendation[J]. arXiv preprint arXiv:2104.00860, 2021.
6. Pang L, Xu J, Ai Q, et al. Setrank: Learning a permutation-invariant ranking model for information retrieval[C]//Proceedings of the 43rd international ACM SIGIR conference on research and development in information retrieval. 2020: 499-508.

往期回顾

1.Flink ClickHouse Sink：生产级高可用写入方案｜得物技术

2.服务拆分之旅：测试过程全揭秘｜得物技术

3.大模型网关：大模型时代的智能交通枢纽｜得物技术

4.从“人治”到“机治”：得物离线数仓发布流水线质量门禁实践

5.AI编程实践：从Claude Code实践到团队协作的优化思考｜得物技术

文 /张卓

关注得物技术，每周一、三更新技术干货

要是觉得文章对你有帮助的话，欢迎评论转发点赞～

未经得物技术许可严禁转载，否则依法追究法律责任。

一、背景与痛点

业务场景

在实时大数据处理场景中，Flink + ClickHouse 的组合被广泛应用于：

• 日志处理： 海量应用日志实时写入分析库。
• 监控分析： 业务指标、APM 数据的实时聚合。

这些场景的共同特点：

• 数据量大：百万级 TPS，峰值可达千万级。
• 写入延迟敏感： 需要秒级可见。
• 数据准确性要求高：不允许数据丢失。
• 多表写入： 不同数据根据分表策略写入不同的表。

开源 Flink ClickHouse Sink 的痛点

Flink 官方提供的 ClickHouse Sink（flink-connector-jdbc）在生产环境中存在以下严重问题：

痛点一：缺乏基于数据量的攒批机制

问题表现：

// Flink 官方 JDBC Sink 的实现
public class JdbcSink<T> extends RichSinkFunction<T> {
    private final int batchSize;  // 固定批次大小
    @Override
    public void invoke(T value, Context context) {
        bufferedValues.add(value);
        if (bufferedValues.size() >= batchSize) {
            // 只能基于记录数攒批，无法基于数据量
            flush();
        }
    }

带来的问题：

1. 内存占用不可控： 100 条 1KB 的日志和 100 条 10MB 的日志占用内存差距 100 倍。
2. OOM 风险高： 大日志记录（如堆栈转储）会迅速撑爆内存。
3. 写入性能差： 无法根据记录大小动态调整批次，导致小记录批次过大浪费网络开销。

痛点二：无法支持动态表结构

问题表现：

// Flink 官方 Sink 只能写入固定表
public class JdbcSink {
    private final String sql;  // 固定的 INSERT SQL
    public JdbcSink(String jdbcUrl, String sql, ...) {
        this.sql = sql;  // 硬编码的表结构
    }
}

带来的问题：

1. 多应用无法隔离： 所有应用的数据写入同一张表，通过特定分表策略区分。
2. 扩展性差： 新增应用需要手动建表，无法动态路由。
3. 性能瓶颈： 单表数据量过大（百亿级），查询和写入性能急剧下降。

痛点三：分布式表写入性能问题

问题表现：

// 大多数生产实现直接写入分布式表
INSERT INTO distributed_table_all VALUES (...)

ClickHouse 分布式表的工作原理：

带来的问题：

1. 网络开销大： 数据需要经过分布式表层转发，延迟增加。
2. 写入性能差： 分布式表增加了路由和转发逻辑，吞吐量降低。
3. 热点问题： 所有数据先到分布式表节点，再转发，造成单点瓶颈。

生产级方案的核心改进

针对以上痛点，本方案提供了以下核心改进：

改进一：基于数据量的攒批机制

public class ClickHouseSinkCounter {
    private Long metaSize;  // 累计数据量（字节）
    public void add(LogModel value) {
        this.values.add(value);
        this.metaSize += value.getMetaSize();  // 累加数据量
    }
}
// 触发条件
private boolean flushCondition(String application) {
    return checkMetaSize(application)  // metaSize >= 10000 字节
        || checkTime(application);     // 或超时 30 秒
}

优势：

• 内存可控： 根据数据量而非记录数攒批。
• 精确控制： 1KB 的记录攒 10000 条 = 10MB，1MB 的记录攒 10 条 = 10MB。
• 避免OOM： 大日志记录不会撑爆内存。

改进二：动态表结构与分片策略

public abstract class ClickHouseShardStrategy<T> {
    public abstract String getTableName(T data);
}
//日志侧实现为应用级分表
public class LogClickHouseShardStrategy extends ClickHouseShardStrategy<String> {
    @Override
    public String getTableName(String application) {
        // 动态路由：order-service → tb_logs_order_service
        return String.format("tb_logs_%s", application);
    }
}

优势：

• 应用隔离： 日志侧内置应用级分表，每个应用独立分表。
• 动态路由： 根据 application 自动路由到目标表。
• 扩展性强： 新增应用无需手动建表（配合 ClickHouse 自动建表）。

改进三：本地表写入 + 动态节点发现

public class ClickHouseLocalWriter extends ClickHouseWriter {
    // 直接写本地表，避免分布式表转发
    private final ConcurrentMap<String, HikariDataSource> dataSourceMap;
    @Override
    public HikariDataSource getNextDataSource(Set<String> exceptionHosts) {
        // 1. 动态获取集群节点列表
        List<String> healthyHosts = getHealthyHosts(exceptionHosts);
        // 2. 随机选择健康节点
        return dataSourceMap.get(healthyHosts.get(random.nextInt(size)));
    }
}

优势：

• 性能提升： 直接写本地表，避免网络转发。
• 高可用： 动态节点发现 + 故障节点剔除。
• 负载均衡： 随机选择 + Shuffle 初始化。

技术方案概览

基于以上改进，本方案提供了以下核心能力：

1. 本地表/分布式表写入： 性能优化与高可用平衡。
2. 分片策略： 按应用维度路由与隔离。
3. 攒批与内存控制： 双触发机制（数据量 + 超时）。
4. 流量控制与限流： 有界队列 + 连接池。
5. 健壮的重试机制： 递归重试 + 故障节点剔除。
6. Checkpoint 语义保证： At-Least-Once 数据一致性。

二、核心架构设计

架构图

核心组件

核心流程

三、本地表 vs 分布式表写入

ClickHouse 表结构说明

ClickHouse 推荐直接写本地表，原因：

1. 写入性能： 避免分布式表的网络分发。
2. 数据一致性： 直接写入目标节点，减少中间环节故障点，比分布式表写入更安全，利于工程化。
3. 负载均衡： 客户端路由实现负载分散。

-- 本地表（实际存储数据）
CREATE TABLE tb_logs_local ON CLUSTER 'default' (
    application String,
    environment String,
    message String,
    log_time DateTime
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(log_time)
ORDER BY (application, log_time);
-- 分布式表（逻辑视图，不存储数据）
CREATE TABLE tb_logs_all ON CLUSTER 'default' AS tb_logs_local
ENGINE = Distributed('default', dw_log, tb_logs_local, cityHash64(application));

HikariCP 连接池配置

// HikariCP 连接池配置
public class ClickHouseDataSourceUtils {
    private static HikariConfig getHikariConfig(DataSourceImpl dataSource) {
        HikariConfig config = new HikariConfig();
        config.setConnectionTimeout(30000L);    // 连接超时 30s
        config.setMaximumPoolSize(20);          // 最大连接数 20
        config.setMinimumIdle(2);               // 最小空闲 2
        config.setDataSource(dataSource);
        return config;
    }
    private static Properties getClickHouseProperties(ClickHouseSinkCommonParams params) {
        Properties props = new Properties();
        props.setProperty("user", params.getUser());
        props.setProperty("password", params.getPassword());
        props.setProperty("database", params.getDatabase());
        props.setProperty("socket_timeout", "180000");      // Socket 超时 3 分钟
        props.setProperty("socket_keepalive", "true");      // 保持连接
        props.setProperty("http_connection_provider", "APACHE_HTTP_CLIENT");
        return props;
    }
}

配置说明：

• maxPoolSize=20：每个 ClickHouse 节点最多 20 个连接。
• minIdle=2：保持 2 个空闲连接，避免频繁创建。
• socket_timeout=180s：Socket 超时 3 分钟，防止长时间查询阻塞。

ClickHouseLocalWriter：动态节点发现

public class ClickHouseLocalWriter extends ClickHouseWriter {
    // 本地节点缓存，按 IP 维护
    private final ConcurrentMap<String, HikariDataSource> dataSourceMap;
    // 动态获取集群本地表节点
    private final ClusterIpsUtils clusterIpsUtils;
    // IP 变更标志（CAS 锁，避免并发更新）
    private static final AtomicBoolean IP_CHANGING = new AtomicBoolean(false);
    @Override
    public HikariDataSource getNextDataSource(Set<String> exceptionHosts) {
        // 1️⃣ 检测集群节点变化（通过 CAS 避免并发更新）
        if (clusterIpsChanged() && IP_CHANGING.compareAndSet(false, true)) {
            try {
                ipChanged(); // 动态更新 dataSourceMap
            } finally {
                IP_CHANGING.set(false);
            }
        }
        // 2️⃣ 获取异常节点列表（从 Redis + APM 实时查询）
        Set<String> exceptIps = clusterIpsUtils.getExceptIps();
        exceptIps.addAll(exceptionHosts);
        // 3️⃣ 过滤健康节点，随机选择
        List<String> healthyHosts = dataSourceMap.keySet().stream()
            .filter(host -> !exceptIps.contains(host))
            .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(healthyHosts)) {
            throw new RuntimeException("Can't get datasource from local cache");
        }
        return dataSourceMap.get(healthyHosts.get(random.nextInt(healthyHosts.size())));
    }
    private void ipChanged() {
        List<String> clusterIps = clusterIpsUtils.getClusterIps();
        // 新增节点：自动创建连接池
        clusterIps.forEach(ip ->
            dataSourceMap.computeIfAbsent(ip, v ->
                createHikariDataSource(ip, port)
            )
        );
        // 移除下线节点：关闭连接池
        dataSourceMap.forEach((ip, ds) -> {
            if (!clusterIps.contains(ip)) {
                dataSourceMap.remove(ip);
                ds.close();
            }
        });
    }
}

核心逻辑：

1. 动态节点发现： 从 system.clusters 查询所有节点。
2. 自动扩缩容： 节点上线自动加入，下线自动剔除。
3. 故障节点剔除： 通过 APM 监控，自动剔除异常节点。
4. 负载均衡： 随机选择健康节点，避免热点。

集群节点动态发现（ClusterIpsUtils）

public class ClusterIpsUtils {
    // 从 system.clusters 查询所有节点
    private static final String QUERY_CLUSTER_IPS =
        "select host_address from system.clusters where cluster = 'default'";
    // LoadingCache：定时刷新节点列表（1 小时）
    private final LoadingCache<String, List<String>> clusterIpsCache =
        CacheBuilder.newBuilder()
            .expireAfterAccess(10, TimeUnit.HOURS)
            .refreshAfterWrite(1, TimeUnit.HOURS)
            .build(CacheLoader.asyncReloading(new CacheLoader<>() {
                @Override
                public List<String> load(String dbName) {
                    return queryClusterIps();  // 定时刷新节点列表
                }
            }));
    // 异常节点缓存（1 分钟刷新）
    private final LoadingCache<String, FlinkExceptIpModel> exceptIpsCache =
        CacheBuilder.newBuilder()
            .refreshAfterWrite(1, TimeUnit.MINUTES)
            .build(CacheLoader.asyncReloading(new CacheLoader<>() {
                @Override
                public FlinkExceptIpModel load(String dbName) {
                    return queryExceptIp();  // 从 Redis + APM 查询异常节点
                }
            }));
}

异常节点监控策略：

• 磁盘使用率 >= 90%： 从 APM 查询 Prometheus 指标，自动加入黑名单。
• HTTP 连接数 >= 50： 连接数过多说明节点压力大，自动加入黑名单。
• 人工配置： 通过 Redis 配置手动剔除节点

数据来源：

1. ClickHouse system.clusters 表： 获取所有集群节点。
2. APM Prometheus 接口： 监控节点健康状态。
3. Redis 缓存： 人工配置的异常节点。

负载均衡优化

public class ClickHouseWriter {
    public <T> ClickHouseWriter(...) {
        // Shuffle：随机打乱数据源顺序
        Collections.shuffle(clickHouseDataSources);
        this.clickHouseDataSources = clickHouseDataSources;
    }
    public HikariDataSource getNextDataSource(Set<String> exceptionHosts) {
        // 轮询 + 随机选择（已 shuffle，避免热点）
        int current = this.currentRandom.getAndIncrement();
        if (current >= clickHouseDataSources.size()) {
            this.currentRandom.set(0);
        }
        return clickHouseDataSources.get(currentRandom.get());
    }
}

优势：

• 初始化时 shuffle，避免所有 writer 同时从第一个节点开始。
• 轮询 + 随机选择，负载分散更均匀。
• 故障节点自动剔除。

四、支持分表策略

分片策略抽象

public abstract class ClickHouseShardStrategy<T> {
    private String tableName;      // 表名模板，如 "tb_log_%s"
    private Integer tableCount;    // 分表数量
    // 根据数据决定目标表名
    public abstract String getTableName(T data);
}

日志分片实现

public class LogClickHouseShardStrategy extends ClickHouseShardStrategy<String> {
    @Override
    public String getTableName(String application) {
        // 表名格式：tb_log_{application}
        // 例如：application = "order-service" -> table = "tb_log_order_service"
        return String.format(
            this.getTableName(),
            application.replace("-", "_").toLowerCase()
        );
    }
}

按表（应用）维度的缓冲区

日志侧维度降级为应用名称维度缓冲区，实则因为按照应用分表，

业务方可使用自身分表策略携带表名元数据，进行表维度缓冲。

public class ClickHouseShardSinkBuffer {
    // 按 application 分组的缓冲区（ConcurrentHashMap 保证并发安全）
    private final ConcurrentHashMap<String, ClickHouseSinkCounter> localValues;
    public void put(LogModel value) {
        String application = value.getApplication();
        // 1️⃣ 检查是否需要 flush
        if (flushCondition(application)) {
            addToQueue(application); // 触发写入
        }
        // 2️⃣ 添加到缓冲区（线程安全的 compute 操作）
        localValues.compute(application, (k, v) -> {
            if (v == null) v = new ClickHouseSinkCounter();
            v.add(value);
            return v;
        });
    }
    private void addToQueue(String application) {
        localValues.computeIfPresent(application, (k, v) -> {
            // 深拷贝并清空（避免并发修改异常）
            List<LogModel> deepCopy = v.copyValuesAndClear();
            // 构造请求 Blank：application + targetTable + values
            String targetTable = shardStrategy.getTableName(application);
            ClickHouseRequestBlank blank = new ClickHouseRequestBlank(deepCopy, application, targetTable);
            // 放入队列
            writer.put(blank);
            return v;
        });
    }
}

核心设计：

• 应用隔离： 每个表（应用）独立的 buffer，互不影响。
• 线程安全： 使用 ConcurrentHashMap.compute()保证并发安全。
• 深拷贝： List.copyOf() 创建不可变副本，避免并发修改。
• 批量清空： 一次性取出所有数据，清空计数器。

五、攒批与内存控制

双触发机制

public class ClickHouseShardSinkBuffer {
    private final int maxFlushBufferSize;  // 最大批次大小（如 10000）
    private final long timeoutMillis;      // 超时时间（如 30s）
    // 触发条件检查（满足任一即触发）
    private boolean flushCondition(String application) {
        return localValues.get(application) != null
            && (checkMetaSize(application) || checkTime(application));
    }
    // 条件1：达到批次大小
    private boolean checkMetaSize(String application) {
        return localValues.get(application).getMetaSize() >= maxFlushBufferSize;
    }
    // 条件2：超时
    private boolean checkTime(String application) {
        long current = System.currentTimeMillis();
        return current - localValues.get(application).getInsertTime() > timeoutMillis;
    }
}

批次大小计算

public class ClickHouseSinkCounter {
    private final List<LogModel> values;
    private Long metaSize; // 累计的 metaSize（字节）
    public void add(LogModel value) {
        this.values.add(value);
        this.metaSize += value.getMetaSize(); // 累加 metaSize
    }
    public List<LogModel> copyValuesAndClear() {
        List<LogModel> logModels = List.copyOf(this.values); // 深拷贝（不可变）
        this.values.clear();
        this.metaSize = 0L;
        this.insertTime = System.currentTimeMillis();
        return logModels;
    }
}

关键点：

• 使用 metaSize（字节数）而非记录数控制批次，内存控制更精确。
• List.copyOf() 创建不可变副本，避免并发修改。
• 清空后重置 insertTime，保证超时触发准确性。

带随机抖动的超时

private final long timeoutMillis;
public ClickHouseShardSinkBuffer(..., int timeoutSec, ...) {
    // 基础超时 + 10% 随机抖动（避免惊群效应）
    this.timeoutMillis = TimeUnit.SECONDS.toMillis(timeoutSec)
                      + new SecureRandom().nextInt((int) (timeoutSec * 0.1 * 1000));
}

目的： 避免多个TM 同时触发 flush，造成写入流量峰值。

配置示例

ClickHouseShardSinkBuffer.Builder
    .aClickHouseSinkBuffer()
    .withTargetTable("single_table")  //单表时，可直接使用指定表名
    .withMaxFlushBufferSize(10000)  // 对应字节数
    .withTimeoutSec(30)              // 30 秒超时
    .withClickHouseShardStrategy(new LogClickHouseShardStrategy("table_prefix_%s", 8))  //分表策略时，使用
    // 分表策略可根据业务实际情况进行扩展
    .build(clickHouseWriter);

六、写入限流与流量控制

有界队列设计

public class ClickHouseWriter {
    // 有界阻塞队列
    private final BlockingQueue<ClickHouseRequestBlank> commonQueue;
    public ClickHouseWriter(ClickHouseSinkCommonParams sinkParams, ...) {
        // 队列最大容量配置（默认 10）
        this.commonQueue = new LinkedBlockingQueue<>(sinkParams.getQueueMaxCapacity());
    }
    public void put(ClickHouseRequestBlank params) {
        unProcessedCounter.incrementAndGet();
        // put() 方法在队列满时会阻塞，实现背压
        commonQueue.put(params);
    }
}

背压传导：

线程池并发控制

public class ClickHouseWriter {
    private final int numWriters; // 写入线程数
    private ExecutorService service;
    private void buildComponents() {
        ThreadFactory threadFactory = ThreadUtil.threadFactory("clickhouse-writer");
        service = Executors.newFixedThreadPool(numWriters, threadFactory);
        // 创建多个 WriterTask 并提交
        for (int i = 0; i < numWriters; i++) {
            WriterTask task = new WriterTask(i, commonQueue, sinkParams, futures, unProcessedCounter);
            service.submit(task);
        }
    }
}

WriterTask 消费逻辑

class WriterTask implements Runnable {
    @Override
    public void run() {
        isWorking = true;
        while (isWorking || !queue.isEmpty()) {
            // poll() 超时返回（100ms），避免无限等待
            ClickHouseRequestBlank blank = queue.poll(100, TimeUnit.MILLISECONDS);
            if (blank != null) {
                // 创建 Future 并设置超时（3 分钟）
                CompletableFuture<Boolean> future = new CompletableFuture<>();
                future.orTimeout(3, TimeUnit.MINUTES);
                futures.add(future);
                try {
                    send(blank, future, new HashSet<>());
                } finally {
                    // final 进行未知异常兜底，防止为捕获异常造成future状态不完成，永久阻塞
                    if (!future.isDone()) {
                        future.completeExceptionally(new RuntimeException("Unknown exception"));
                    }
                    queueCounter.decrementAndGet();
                }
            }
        }
    }
}

配置参数

七、重试机制与超时控制

Future 超时控制

public class ClickHouseWriter {
    private final int numWriters; // 写入线程数
    private ExecutorService service;
    private void buildComponents() {
        ThreadFactory threadFactory = ThreadUtil.threadFactory("clickhouse-writer");
        service = Executors.newFixedThreadPool(numWriters, threadFactory);
        // 创建多个 WriterTask 并提交
        for (int i = 0; i < numWriters; i++) {
            WriterTask task = new WriterTask(i, commonQueue, sinkParams, futures, unProcessedCounter);
            service.submit(task);
        }
    }
}

超时策略：

• Future 超时： 3 分钟（orTimeout）。
• Socket 超时： 3 分钟（socket_timeout=180000）。
• 连接超时： 30 秒（connectionTimeout=30000）。

重试逻辑

class WriterTask implements Runnable {
    @Override
    public void run() {
        isWorking = true;
        while (isWorking || !queue.isEmpty()) {
            // poll() 超时返回（100ms），避免无限等待
            ClickHouseRequestBlank blank = queue.poll(100, TimeUnit.MILLISECONDS);
            if (blank != null) {
                // 创建 Future 并设置超时（3 分钟）
                CompletableFuture<Boolean> future = new CompletableFuture<>();
                future.orTimeout(3, TimeUnit.MINUTES);
                futures.add(future);
                try {
                    send(blank, future, new HashSet<>());
                } finally {
                    // final 进行未知异常兜底，防止为捕获异常造成future状态不完成，永久阻塞
                    if (!future.isDone()) {
                        future.completeExceptionally(new RuntimeException("Unknown exception"));
                    }
                    queueCounter.decrementAndGet();
                }
            }
        }
    }
}

重试控制逻辑

private void handleUnsuccessfulResponse(..., Set<String> exceptHosts) {
    // 检查 Future 是否已完成（避免重复完成）
    if (future.isDone()) {
        return;
    }
    if (attemptCounter >= maxRetries) {
        // 达到最大重试次数，标记失败
        future.completeExceptionally(new RuntimeException("Max retries exceeded"));
    } else {
        // 递归重试
        requestBlank.incrementCounter();
        send(requestBlank, future, exceptHosts); // 递归调用，排除失败节点
    }
}

重试策略：

• 递归重试： 失败后递归调用，直到成功或达到最大次数。
• 异常节点隔离： 每次重试时排除失败的节点（exceptHosts）。
• 超时控制： Future 超时（3 分钟）防止永久阻塞。

为什么递归重试是更好的选择

递归重试（当前实现）

队列重试（假设方案）

保证一致性

  // ClickHouseWriter.java:139-158
  while (!futures.isEmpty() || unProcessedCounter.get() > 0) {
      CompletableFuture<Void> future = FutureUtil.allOf(futures);
      future.get(3, TimeUnit.MINUTES);  // 阻塞直到全部完成
  }

• Checkpoint 时所有数据要么全部成功，要么全部失败。
• 重启后不会有部分数据重复的问题。

简单可靠

• 代码逻辑清晰。
• 对于队列重试且不重复，需要复杂的二阶段提交（这里暂不展开），大幅增加代码复杂度。

性能可接受

class WriterTask implements Runnable {
    @Override
    public void run() {
        while (isWorking || !queue.isEmpty()) {
            ClickHouseRequestBlank blank = queue.poll(100, TimeUnit.MILLISECONDS);
            if (blank != null) {
                // 创建 Future 并设置 3 分钟超时
                CompletableFuture<Boolean> future = new CompletableFuture<>();
                future.orTimeout(3, TimeUnit.MINUTES); // 防止永久阻塞
                futures.add(future);
                try {
                    send(blank, future, new HashSet<>());
                } finally {
                    if (!future.isDone()) {
                        future.completeExceptionally(new RuntimeException("Timeout"));
                    }
                    queueCounter.decrementAndGet();
                }
            }
        }
    }
}

• 虽然阻塞，但有超时保护。
• ClickHouse 写入通常很快（秒级）。
• 网络故障时重试也合理。

避开故障节点

  // ClickHouseWriter.java:259-260
  HikariDataSource dataSource = getNextDataSource(exceptHosts);

• 递归时可以传递 exceptHosts。
• 自动避开失败的节点。
• 提高成功率。

异常节点剔除

// 特殊错误码列表（自动加入黑名单）
private final List<Integer> ignoreHostCodes = Arrays.asList(210, 1002);
public HikariDataSource getNextDataSource(Set<String> exceptionHosts) {
    if (CollectionUtils.isNotEmpty(exceptionHosts)) {
        // 过滤异常节点
        List<HikariDataSource> healthyHosts = clickHouseDataSources.stream()
            .filter(ds -> !exceptionHosts.contains(getHostFromUrl(ds)))
            .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(healthyHosts)) {
            return null; // 所有节点都异常
        }
        return healthyHosts.get(random.nextInt(healthyHosts.size()));
    }
    // 正常轮询（已 shuffle，避免热点）
    return clickHouseDataSources.get(currentRandom.getAndIncrement() % size);
}

故障节点剔除策略：

1. 错误码 210（网络异常）： 自动加入黑名单。
2. 错误码 1002（连接池异常）： 自动加入黑名单。
3. APM 监控： 磁盘 >= 90%、HTTP 连接 >= 50 的节点。
4. 手动配置： 通过 Redis 配置剔除。

恢复机制：

• LoadingCache 定时刷新（1 分钟）。
• 节点恢复健康后自动从黑名单移除。

重试流程图

八、异常处理模式

两种 Sink 模式

public Sink buildSink(String targetTable, String targetCount, int maxBufferSize) {
    IClickHouseSinkBuffer buffer = ClickHouseShardSinkBuffer.Builder
        .aClickHouseSinkBuffer()
        .withTargetTable(targetTable)
        .withMaxFlushBufferSize(maxBufferSize)
        .withClickHouseShardStrategy(new LogClickHouseShardStrategy(targetTable, count))
        .build(clickHouseWriter);
    // 根据配置选择模式
    if (ignoringClickHouseSendingExceptionEnabled) {
        return new UnexceptionableSink(buffer);  // 忽略异常
    } else {
        return new ExceptionsThrowableSink(buffer); // 抛出异常
    }
}

UnexceptionableSink（忽略异常 - At-Most-Once）

public class UnexceptionableSink implements Sink<LogModel> {
    private final IClickHouseSinkBuffer<LogModel> buffer;
    @Override
    public void put(LogModel message) {
        buffer.put(message);  // 不检查 Future 状态
    }
    @Override
    public void flush() {
        buffer.flush();
    }
}

适用场景：

• 允许部分数据丢失。
• 不希望因写入异常导致任务失败。
• 对数据准确性要求不高（如日志统计）。

语义保证：At-Most-Once（最多一次）

ExceptionsThrowableSink（抛出异常 - At-Least-Once）

public class ExceptionsThrowableSink implements Sink<LogModel> {
    private final IClickHouseSinkBuffer<LogModel> buffer;
    @Override
    public void put(LogModel message) throws ExecutionException, InterruptedException {
        buffer.put(message);
        // 每次写入都检查 Future 状态
        buffer.assertFuturesNotFailedYet();
    }
    @Override
    public void flush() throws ExecutionException, InterruptedException {
        buffer.flush();
    }
}

Future 状态检查：

public void assertFuturesNotFailedYet() throws ExecutionException, InterruptedException {
    CompletableFuture<Void> future = FutureUtil.allOf(futures);
    // 非阻塞检查
    if (future.isCompletedExceptionally()) {
        logger.error("There is something wrong with the future. exist sink now");
        future.get(); // 抛出异常，导致 Flink 任务失败
    }
}

适用场景：

• 数据准确性要求高。
• 需要保证所有数据写入成功。
• 异常时希望 Flink 任务失败并重启。

语义保证：At-Least-Once（至少一次）

Future 清理策略与并发控制

定时检查器

public class ClickHouseSinkScheduledCheckerAndCleaner {
    private final ScheduledExecutorService scheduledExecutorService;
    private final List<CompletableFuture<Boolean>> futures;
    // ⚠️ volatile 保证多线程可见性（关键并发控制点）
    private volatile boolean isFlushing = false;
    public ClickHouseSinkScheduledCheckerAndCleaner(...) {
        // 单线程定时执行器
        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(factory);
        // 定时执行清理任务（每隔 checkTimeout 秒，默认 30 秒）
        scheduledExecutorService.scheduleWithFixedDelay(getTask(), ...);
    }
    private Runnable getTask() {
        return () -> {
            synchronized (this) {
                //  关键：检查是否正在 flush，避免并发冲突
                if (isFlushing) {
                    return; // Checkpoint 期间暂停清理
                }
                // 1️⃣ 清理已完成的 Future
                futures.removeIf(filter);
                // 2️⃣ 触发所有 Buffer 的 flush（检查是否需要写入）
                clickHouseSinkBuffers.forEach(IClickHouseSinkBuffer::tryAddToQueue);
            }
        };
    }
    // Checkpoint flush 前调用（暂停 cleaner）
    public synchronized void beforeFlush() {
        isFlushing = true;
    }
    // Checkpoint flush 后调用（恢复 cleaner）
    public synchronized void afterFlush() {
        isFlushing = false;
    }
}

核心设计：

• volatile boolean isFlushing： 标志位，协调 cleaner 与 checkpoint 线程。
• synchronized (this)： 保证原子性，避免并发冲突。
• 单线程执行器： 避免 cleaner 内部并发问题。

并发控制机制

问题场景：

时间轴冲突：
T1: Cleaner 线程正在执行 tryAddToQueue()
T2: Checkpoint 触发，调用 sink.flush()
T3: Cleaner 同时也在执行 tryAddToQueue()
    ├─ 可能导致：数据重复写入
    ├─ 可能导致：Buffer 清空顺序混乱
    └─ 可能导致：Future 状态不一致

解决方案：

// ClickHouseSinkManager.flush()
public void flush() {
    // 1️⃣ 暂停定时清理任务（设置标志）
    clickHouseSinkScheduledCheckerAndCleaner.beforeFlush(); // isFlushing = true
    try {
        // 2️⃣ 执行 flush（此时 cleaner 线程会跳过执行）
        clickHouseWriter.waitUntilAllFuturesDone(false, false);
    } finally {
        // 3️⃣ 恢复定时清理任务
        clickHouseSinkScheduledCheckerAndCleaner.afterFlush(); // isFlushing = false
    }
}

并发控制流程：

关键设计点：

1. volatile 保证可见性： isFlushing 使用 volatile，确保多线程间的可见性。
2. synchronized 保证原子性： getTask() 整个方法体使用 synchronized (this)。
3. 标志位协调： 通过 isFlushing 标志实现两个线程间的协调。
4. finally 确保恢复： 即使 waitUntilAllFuturesDone() 异常，也会在 finally 中恢复 cleaner。

避免的并发问题：

• 数据重复写入： Cleaner 和 Checkpoint 同时 flush。
• Buffer 状态不一致： 一边清空一边写入。
• Future 清理冲突： 正在使用的 Future 被清理。

性能影响：

• Checkpoint flush 期间，cleaner 暂停执行（通常 1-3 秒）。
• Cleaner 跳过的周期会在下次正常执行时补偿。
• 对整体吞吐影响极小（cleaner 间隔通常 30 秒）。

九、Checkpoint 语义保证

为什么 Checkpoint 时必须 Flush？

不 Flush 的后果

不Flush导致数据永久丢失

正确做法

@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    logger.info("start doing snapshot. flush sink to ck");
    // 1. 先 flush buffer（将内存数据写入 ClickHouse）
    if (sink != null) {
        sink.flush();
    }
    // 2. 等待所有写入完成
    if (sinkManager != null && !sinkManager.isClosed()) {
        sinkManager.flush();
    }
    // 此时 Checkpoint 才能标记为成功
    logger.info("doing snapshot. flush sink to ck");
}

Flush 实现与并发协调

public class ClickHouseSinkManager {
    public void flush() {
        //  步骤1：暂停定时清理任务
        clickHouseSinkScheduledCheckerAndCleaner.beforeFlush(); // isFlushing = true
        try {
            //  步骤2：执行 buffer flush + 等待所有写入完成
            clickHouseWriter.waitUntilAllFuturesDone(false, false);
        } finally {
            //  步骤3：恢复定时清理任务（finally 确保执行）
            clickHouseSinkScheduledCheckerAndCleaner.afterFlush(); // isFlushing = false
        }
    }
}

并发协调详解：

// cleaner 线程执行流程
synchronized (this) {
    if (isFlushing) {
        return; // Checkpoint 期间跳过本次执行
    }
    // 正常执行：清理已完成的 Future + 触发 Buffer flush
    futures.removeIf(filter);
    buffers.forEach(Buffer::tryAddToQueue);
}

关键点：

• volatile 可见性： isFlushing 使用 volatile 确保 cleaner 线程立即看到状态变化。
• synchronized互斥： getTask()方法体使用 synchronized (this) 确保原子性。
• 标志位协调： 通过 beforeFlush() / afterFlush() 管理标志位。
• finally 保证恢复： 即使 flush 异常，也会在 finally 中恢复 cleaner。

等待所有 Future 完成

public synchronized void waitUntilAllFuturesDone(boolean stopWriters, boolean clearFutures) {
    try {
        // 循环等待：直到所有 Future 完成 + 队列清空
        while (!futures.isEmpty() || unProcessedCounter.get() > 0) {
            CompletableFuture<Void> all = FutureUtil.allOf(futures);
            // 最多等待 3 分钟（与 Future 超时一致）
            all.get(3, TimeUnit.MINUTES);
            // 移除已完成的 Future（非异常）
            futures.removeIf(f -> f.isDone() && !f.isCompletedExceptionally());
            // 检查是否有异常 Future
            if (anyFutureFailed()) {
                break; // 有异常则退出
            }
        }
    } finally {
        if (stopWriters) stopWriters();
        if (clearFutures) futures.clear();
    }
}

关键逻辑：

• 循环等待直到所有 Future 完成 + 队列清空。
• 超时 3 分钟（与 Future 超时一致）。
• 移除已完成的非异常 Future。
• 有异常时退出循环。

三种 Flush 触发方式对比

Checkpoint 参数配置

// Checkpoint 配置建议
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 启用 Checkpoint（间隔 1 分钟）
env.enableCheckpointing(60000);
// Checkpoint 超时（必须大于 Future 超时 + 重试时间）
// 建议：CheckpointTimeout > FutureTimeout * MaxRetries
env.getCheckpointConfig().setCheckpointTimeout(600000); // 10 分钟
// 一致性模式
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 最小间隔（避免过于频繁）
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); // 30 秒
// 最大并发 Checkpoint 数
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

语义保证

推荐配置：

生产环境： 使用 ExceptionsThrowableSink + Checkpoint。

允许部分丢失： 使用 UnexceptionableSink。

十、最佳实践与调优

生产配置

// ========== ClickHouse 连接参数 ==========
clickhouse.sink.target-table = tb_logs_local
clickhouse.sink.max-buffer-size = 104857600        // 批次大小
clickhouse.sink.table-count = 0                // 0 表示不分表
// ========== 写入性能参数 ==========
clickhouse.sink.num-writers = 10               // 写入线程数
clickhouse.sink.queue-max-capacity = 10        // 队列容量
clickhouse.sink.timeout-sec = 30               // flush 超时
clickhouse.sink.retries = 10                   // 最大重试次数
clickhouse.sink.check.timeout-sec = 30         // 定时检查间隔
// ========== 异常处理参数 ==========
clickhouse.sink.ignoring-clickhouse-sending-exception-enabled = false
clickhouse.sink.local-address-enabled = true   // 启用本地表写入
// ========== ClickHouse 集群配置 ==========
clickhouse.access.hosts = 192.168.1.1:8123,192.168.1.2:8123,192.168.1.3:8123
clickhouse.access.user = default
clickhouse.access.password = ***
clickhouse.access.database = dw_xx_xx
clickhouse.access.cluster = default
// ========== HikariCP 连接池配置 ==========
connectionTimeout = 30000                      // 连接超时 30s
maximumPoolSize = 20                           // 最大连接数 20
minimumIdle = 2                                // 最小空闲 2
socket_timeout = 180000                        // Socket 超时 3mi

性能调优

故障排查

十一、总结

本文深入分析了 Flink ClickHouse Sink 的实现方案，核心亮点包括：

技术亮点

• 连接池选型： 使用 HikariCP，性能优异，连接管理可靠。
• Future 超时控制： orTimeout(3min) 防止永久阻塞。
• 显式资源管理： Connection 和 PreparedStatement 显式关闭，防止连接泄漏。
• 负载均衡优化： Shuffle 初始化 + 轮询选择，避免热点。
• 异常处理增强： future.isDone() 检查，避免重复完成。
• 本地表写入： 动态节点发现 + 故障剔除，写入性能提升。
• 分片策略： 按表（应用）维度路由，独立缓冲和隔离。
• 攒批优化： 双触发机制（大小 + 超时）+ 随机抖动。
• 流量控制： 有界队列 + 线程池，实现背压。
• 健壮重试： 递归重试 + 异常节点剔除 + 最大重试限制。

Checkpoint 语义

• At-Least-Once： ExceptionsThrowableSink + Checkpoint。
• At-Most-Once： UnexceptionableSink。
• Exactly-Once： 需要配合 ClickHouse 事务（未实现）。

生产建议

1. 必须： Checkpoint 时 flush，否则会丢数据。
2. 推荐： 使用 HikariCP + 本地表写入。
3. 推荐： 配置合理的超时（Future < Socket < Checkpoint）。
4. 推荐： 监控队列大小、Future 失败率、重试次数。

该方案已在生产环境大规模验证，能够稳定支撑百万级 TPS 的日志写入场景。

往期回顾

1.服务拆分之旅：测试过程全揭秘｜得物技术

2.大模型网关：大模型时代的智能交通枢纽｜得物技术

3.从“人治”到“机治”：得物离线数仓发布流水线质量门禁实践

4.AI编程实践：从Claude Code实践到团队协作的优化思考｜得物技术

5.入选AAAI-PerFM｜得物社区推荐之基于大语言模型的新颖性推荐算法

文 /虚白

关注得物技术，每周更新技术干货

要是觉得文章对你有帮助的话，欢迎评论转发点赞～

未经得物技术许可严禁转载，否则依法追究法律责任。

一、引言

代码越写越多怎么办？在线等挺急的！ Bidding-interface服务代码库代码量已经达到100w行！！

Bidding-interface应用是出价域核心应用之一，主要面向B端商家。跟商家后台有关的出价功能都围绕其展开。是目前出价域代码量最多的服务。

随着出价业务最近几年来的快速发展，出价服务承接的流量虽然都是围绕卖家出价，但是已远远超过卖家出价功能范围。业务的快速迭代而频繁变更给出价核心链路高可用、高性能都带来了巨大的风险。

经总结有如下几个痛点：

• 核心出价链路未隔离：

  出价链路各子业务模块间代码有不同程度的耦合，迭代开发可扩展性差，往往会侵入到出价主流程代码的改动。每个子模块缺乏独立的封装，而且存在大量重复的代码，每次业务规则调整，需要改动多处，容易出现漏改漏测的问题。

• 大单体&功能模块定义混乱：

  历史原因上层业务层代码缺乏抽象，代码无法实现复用，需求开发代码量大，导致需求估时偏高，经常出现20+人日的大需求，需求开发中又写出大量重复代码，导致出价服务代码库快速膨胀，应用启动耗时过长，恶性循环。

• B/C端链路未隔离：

  B端卖家出价链路流量与C端价格业务场景链路流量没有完全隔离，由于历史原因，有些B端出价链路接口代码还存在于price应用中，偶尔B端需求开发会对C端应用做代码变更。存在一定的代码管控和应用权限管控成本。

• 发布效率影响：

  代码量庞大，导致编译速度缓慢。代码过多，类的依赖关系更为复杂，持续迭代逐步加大编译成本，随着持续迭代，新的代码逻辑 ，引入更多jar 依赖，间接导致项目部署时长变长蓝绿发布和紧急问题处理时长显著增加；同时由于编译与部署时间长，直接影响开发人员在日常迭代中的效率（自测，debug，部署）。

• 业务抽象&分层不合理：

  历史原因出价基础能力领域不明确，出价底层和业务层分层模糊，业务层代码和出价底层代码耦合严重，出价底层能力缺乏抽象，上层业务扩展需求频繁改动出价底层能力代码。给出价核心链路代码质量把控带来较高的成本， 每次上线变更也带来一定的风险。

以上，对于Bidding服务的拆分和治理，已经箭在弦上不得不发。否则，持续的迭代会继续恶化服务的上述问题。

经过前期慎重的筹备，设计，排期，拆分，和测试。目前Bidding应用经过四期的拆分节奏，已经马上要接近尾声了。服务被拆分成三个全新的应用，目前在小流量灰度放量中。

本次拆分涉及：1000+Dubbo接口，300+个HTTP接口，200+ MQ消息，100+个TOC任务，10+个 DJob任务。

本人是出价域测试一枚，参与了一期-四期的拆分测试工作。

项目在全组研发+测试的ALL IN投入下，已接近尾声。值此之际输出一篇文章，从测试视角复盘下，Bidding服务的拆分与治理，也全过程揭秘下出价域内的拆分测试过程。

二、服务拆分的原则

首先，在细节性介绍Bidding拆分之前。先过大概过一下服务拆分原则：

• 单一职责原则 (SRP)：  每个服务应该只负责一项特定的业务功能，避免功能混杂。

• 高内聚、低耦合：  服务内部高度内聚，服务之间松耦合，尽量减少服务之间的依赖关系。

• 业务能力导向：  根据业务领域和功能边界进行服务拆分，确保每个服务都代表一个完整的业务能力。

拆分原则之下，还有不同的策略可以采纳：基于业务能力拆分、基于领域驱动设计 (DDD) 拆分、基于数据拆分等等。同时，拆分时应该注意：避免过度拆分、考虑服务之间的通信成本、设计合理的 API 接口。

服务拆分是微服务架构设计的关键步骤，需要根据具体的业务场景和团队情况进行综合考虑。合理的服务拆分可以提高系统的灵活性、可扩展性和可维护性，而不合理的服务拆分则会带来一系列问题。

三、Bidding服务拆分的设计

如引言介绍过。Bidding服务被拆分出三个新的应用，同时保留bidding应用本身。目前共拆分成四个应用：Bidding-foundtion，Bidding-interface，Bidding-operation和Bidding-biz。详情如下：

• 出价基础服务-Bidding-foundation：

出价基础服务，对出价基础能力抽象，出价领域能力封装，基础能力沉淀。

• 出价服务-Bidding-interfaces：

商家端出价，提供出价基础能力和出价工具，提供商家在各端出价链路能力，重点保障商家出价基础功能和出价体验。

• 出价运营服务-Bidding-operation：

出价运营，重点支撑运营对出价业务相关规则的维护以及平台其他域业务变更对出价域数据变更的业务处理：

1. 出价管理相关配置：出价规则配置、指定卖家规则管理、出价应急隐藏/下线管理工具等；
2. 业务大任务：包括控价生效/失效，商研鉴别能力变更，商家直发资质变更，品牌方出价资质变更等大任务执行。

• 业务扩展服务-Bidding-biz：

更多业务场景扩展，侧重业务场景的灵活扩展，可拆出的现有业务范围：国补采购单出价，空中成单业务，活动出价，直播出价，现订现采业务，预约抢购，新品上线预出价，入仓预出价。

应用拆分前后流量分布情况：

四、Bidding拆分的节奏和目标收益

服务拆分是项大工程，对目前的线上质量存在极大的挑战。合理的排期和拆分计划是重点，可预期的收益目标是灵魂。

经过前期充分调研和规划。Bidding拆分被分成了四期，每期推进一个新应用。并按如下六大步进行：

Bidding拆分目标

• 解决Bidding大单体问题： 对Bidding应用进行合理规划，完成代码和应用拆分，解决一直以来Bidding大单体提供的服务多而混乱，维护成本高，应用编译部署慢，发布效率低等等问题。
• 核心链路隔离&提升稳定性： 明确出价基础能力，对出价基础能力下沉，出价基础能力代码拆分出独立的代码库，并且部署在独立的新应用中，实现出价核心链路隔离，提升出价核心链路稳定性。
• 提升迭代需求开发效率： 完成业务层代码抽象，业务层做组件化配置化，实现业务层抽象复用，降低版本迭代需求开发成本。
• 实现出价业务应用合理规划： 各服务定位、职能明确，分层抽象合理，更好服务于企/个商家、不同业务线运营等不同角色业务推进。

预期的拆分收益

• 出价服务应用结构优化：

  完成对Bidding大单体应用合理规划拆分，向下沉淀出出价基础服务应用层，降低出价基础能力维护成功；向上抽离出业务扩展应用层，能够实现上层业务的灵活扩展；同时把面向平台运营和面向卖家出价的能力独立维护；在代码库和应用层面隔离，有效减少版本迭代业务需求开发变更对应用的影响面，降低应用和代码库的维护成本。

• 完成业务层整体设计，业务层抽象复用，业务层做组件化配置化，提升版本迭代需求开发效率，降低版本迭代需求开发成本：

  按业务类型对业务代码进行分类，统一设计方案，提高代码复用性，支持业务场景变化时快速扩展，以引导降价为例，当有类似降价换流量/降价换销量新的降价场景需求时，可以快速上线，类似情况每个需求可以减少10-20人日开发工作量。

• 代码质量提升 ：

  通过拆分出价基础服务和对出价流程代码做重构，将出价基础底层能力代码与上层业务层代码解耦，降低代码复杂度，降低代码冲突和维护难度，从而提高整体代码质量和可维护性。

• 开发效率提升 ：

  1. 缩短应用部署时间： 治理后的出价服务将加快编译和部署速度，缩短Bidding-interfaces应用发布(编译+部署)时间 由12分钟降低到6分钟，从而显著提升开发人员的工作效率，减少自测、调试和部署所需的时间。以Bidding服务T1环境目前一个月编译部署至少1500次计算，每个月可以节约150h应用发布时间。
  2. 提升问题定位效率： 出价基础服务层与上层业务逻辑层代码库&应用分开后，排查定位开发过程中遇到的问题和线上问题时可以有效缩小代码范围，快速定位问题代码位置。

五、测试计划设计

服务拆分的前期，研发团队投入了大量的心血。现在代码终于提测了，进入我们的测试环节：

为了能收获更好的质量效果，同时也为了不同研发、测试同学的分工。我们需要细化到最细粒度，即接口维度整理出一份详细的文档。基于此文档的基础，我们确定工作量和人员排期：

如本迭代，我们投入4位研发同学，2位测试同学。完成该200个Dubbo接口和100个HTTP接口，以及20个Topic迁移。对应的提测接口，标记上负责的研发、测试、测试进度、接口详细信息等内容。

基于该文档的基础上，我们的工作清晰而明确。一个大型的服务拆分，也变成了一步一步的里程碑任务。

接下来给大家看一下，关于Bidding拆分。我们团队整体的测试计划，我们一共设计了五道流程。

• 第一关：自测接口对比：

  每批次拆分接口提测前，研发同学必须完成接口自测。基于新旧接口返回结果对比验证。验证通过后标记在文档中，再进入测试流程。

  对于拆分项目，自测卡的相对更加严格。由于仅做接口迁移，逻辑无变更，自测也更加容易开展。由研发同学做好接口自测，可以避免提测后新接口不通的低级问题。提高项目进度。

  在这个环节中。偶尔遇见自测不充分、新接口参数传丢、新Topic未配置等问题。（三期、四期测试中，我们加强了对研发自测的要求）。

• 第二关：测试功能回归

  这一步骤基本属于测试的人工验证，同时重点需关注写接口数据验证。

  回归时要测的细致。每个接口，测试同学进行合理评估。尽量针对接口主流程，进行细致功能回归。由于迁移的接口数量多，历史逻辑重。一方面在接口测试任务分配时，要尽量选择对该业务熟悉的同学。另一方面，承接的同学也有做好历史逻辑梳理。尽量不要产生漏测造成的问题。

  该步骤测出的问题五花八门。另外由于Bidding拆分成多个新服务。两个新服务经常彼此间调用会出现问题。比如二期Bidding-foundation迁移完成后，Bidding-operation的接口在迁移时，依赖接口需要从Bidding替换成foundation的接口。

  灰度打开情况下，调用新接口报错仍然走老逻辑。（测试时，需要关注trace中是否走了新应用）。

• 第三关：自动化用例

  出价域内沉淀了比较完善的接口自动化用例。在人工测试时，测试同学可以借助自动化能力，完成对迁移接口的回归功能验证。

  同时在发布前天，组内会特地多跑一轮全量自动化。一次是迁移接口开关全部打开，一次是迁移接口开关全部关闭即正常的自动化回归。然后全员进行排错。

  全量的自动化用例执行，对迁移接口问题拦截，有比较好的效果。因为会有一些功能点，人工测试时关联功能未考虑到，但在接口自动化覆盖下无所遁形。

• 第四关：流量回放

  在拆分接口开关打开的情况下，在预发环境进行流量回放。

  线上录制流量的数据往往更加复杂，经常会测出一些意料之外的问题。

  迭代过程中，我们组内仍然会在沿用两次回放。迁移接口开关打开后回放一次，开关关闭后回放一次。（跟发布配置保持一致）。

• 第五关：灰度过程中，关闭接口开关，功能回滚

  为保证线上生产质量，在迁移接口小流量灰度过程中。我们持续监测线上问题告警群。

  以上，就是出价域测试团队，针对服务拆分的测试流程。同时遵循可回滚的发布标准，拆分接口做了非常完善的灰度功能。下一段落进行介绍。

六、各流量类型灰度切量方案

出价流程切新应用灰度控制从几个维度控制：总开关，出价类型范围，channel范围，source范围，bidSource范围，uid白名单&uid百分比(0-10000)：

• 灰度策略

• 支持 接口维度 ，按照百分比进行灰度切流；

• 支持一键回切；

Dubbo接口、HTTP接口、TOC任务迁移、DMQ消息迁移分别配有不同的灰度策略。

七、结语

拆分的过程中，伴随着很多迭代需求的开发。为了提高迁移效率，我们会在需求排期后，并行处理迭代功能相关的接口，把服务拆分和迭代需求一起完成掉。

目前，我们的拆分已经进入尾声。迭代发布后，整体的技术项目就结束了。灰度节奏在按预期节奏进行~

值得一提的是，目前我们的流量迁移仍处于第一阶段，即拆分应用出价域内灰度迁移，上游不感知。目前所有的流量仍然通过bidding服务接口进行转发。后续第二阶段，灰度验证完成后，需要进行上游接口替换，流量直接请求拆分后的应用。

往期回顾

1.大模型网关：大模型时代的智能交通枢纽｜得物技术

2.从“人治”到“机治”：得物离线数仓发布流水线质量门禁实践

3.AI编程实践：从Claude Code实践到团队协作的优化思考｜得物技术

4.入选AAAI-PerFM｜得物社区推荐之基于大语言模型的新颖性推荐算法

5.Galaxy比数平台功能介绍及实现原理｜得物技术

文 /寇森

关注得物技术，每周一、三更新技术干货

要是觉得文章对你有帮助的话，欢迎评论转发点赞～

未经得物技术许可严禁转载，否则依法追究法律责任。