关于java:vivo-评论中台的流量及数据隔离实践

vivo 评论中台通过提供评论发表、点赞、举报、自定义评论排序等通用能力，帮忙前台业务疾速搭建评论性能并提供评论经营能力，防止了前台业务的反复建设和数据孤岛问题。目前已有 vivo 短视频、vivo 浏览器、负一屏、vivo 商城等 10+ 业务接入。这些业务的流量大小和稳定范畴不同，如何保障各前台业务的高可用，防止因为某个业务的流量暴增导致其余业务的不可用？所有业务的评论数据都交由中台存储，他们的数据量大小不同、db 压力不同，作为中台，应该如何隔离各个业务的数据，保障整个中台零碎的高可用？

本文将和大家一起分享下 vivo 评论中台的解决方案，次要是从流量隔离和数据隔离两局部进行了解决。

vivo 浏览器业务亿级日活，实时热点新闻全网 push，对于这类用户量大、流量大的重要业务，咱们提供了独自的集群为他们提供服务，防止受到其余业务的影响。

vivo 评论中台是通过 Dubbo 接口对外提供服务，咱们通过 Dubbo 标签路由的形式对整个服务集群做了逻辑上的划分，一次 Dubbo 调用可能依据申请携带的 tag 标签智能地抉择对应 tag 的服务提供者进行调用。如下图所示：

1）provider 打标签：目前有两种形式能够实现实例分组，别离是动静规定打标和动态规定打标，其中动静规定相较于动态规定优先级更高，而当两种规定同时存在且呈现抵触时，将以动静规定为准。公司外部的运维零碎很好的反对了动静打标，通过对指定 ip 的机器打标即可 (非 docker 容器，机器 ip 是固定的)。

2）前台 consumer 指定服务标签：发动申请时设置，如下；

前台指定中台的路由标签

RpcContext.getContext().setAttachment(Constants.REQUEST_TAG_KEY,"browser");

申请标签的作用域为每一次 invocation，只须要在调用评论中台服务前设置标签即可，前台业务调用其余业务的 provider 并不受该路由标签的影响。

大流量的业务咱们通过独自的集群隔离进来了。然而独立部署集群老本高，不能为每个前台业务都独立部署一套集群。大部分状况下多个业务还是须要共用一套集群的，那么共用集群的服务遇到了突发流量如何解决呢？没错，限流呗！然而目前很多限流都是一刀切的形式对接口整体 QPS 做限流，这样的话某一前台业务的流量暴增会导致所有前台业务的申请都被限流。

这就须要多租户限流退场了（这里的一个租户能够了解为一个前台业务），反对对同一接口不同租户的流量进行限流解决，成果如下图：

实现过程 ：

咱们应用 sentinel 的热点参数限流个性，应用业务身份编码作为热点参数，为各业务配置不同的流控大小。

那么何为热点参数限流？首先得说下什么是热点，热点即常常拜访的数据。很多时候咱们心愿统计某个热点数据中拜访频次最高的 Top n 数据，并对其拜访进行限度。比方：

商品 ID 为参数，统计一段时间内最常购买的商品 ID 并进行限度。
用户 ID 为参数，针对一段时间内频繁拜访的用户 ID 进行限度。

热点参数限流会统计传入参数中的热点参数，并依据配置的限流阈值与模式，对蕴含热点参数的资源调用进行限流。热点参数限流能够看做是一种非凡的流量管制，仅对蕴含热点参数的资源调用失效。Sentinel 利用 LRU 策略统计最近最常拜访的热点参数，联合令牌桶算法来进行参数级别的流控。下图为评论场景示例：

应用 Sentinel 来进行资源爱护，次要分为几个步骤：定义资源、定义规定、规定失效解决。

1）定义资源 ：

在这里能够了解为各个中台 API 接口门路。

2）定义规定 ：

Sentienl 反对规定很多 QPS 流控、自适应限流、热点参数限流、集群限流等等，这里咱们用的是单机热点参数限流。

热点参数限流配置

{"resource": "com.vivo.internet.comment.facade.comment.CommentFacade:comment(com.vivo.internet.comment.facade.comment.dto.CommentRequestDto)", // 须要限流的接口
    "grade": 1, // QPS 限流模式
    "count": 3000, // 接口默认限流大小 3000
    "clusterMode": false, // 单机模式
    "paramFieldName": "clientCode", // 指定热点参数名即业务方编码字段，这里是咱们对 sentinel 组件做了优化，减少了该配置属性，用来指定参数对象的属性名作为热点参数 key
    "paramFlowItemList": [ // 热点参数限流规定
        {
            "object": "vivo-community", // 当 clientCode 为该值时，匹配该限流规定
            "count": 1000,   // 限流大小为 1000
            "classType": "java.lang.String"
        },
        {
            "object": "vivo-shop", // 当 clientCode 为该值时，匹配该限流规定
            "count": 2000, // 限流大小为 2000
            "classType": "java.lang.String"
        }
    ]
}

3）规定失效解决 ：

当触发了限流规定后 sentinel 会抛出 ParamFlowException 异样，间接将异样抛给前台业务去解决是不优雅的。sentinel 给咱们提供了对立的异样回调解决入口 DubboAdapterGlobalConfig，反对咱们将异样转换为业务自定义后果返回。

自定义限流返回后果；

DubboAdapterGlobalConfig.setProviderFallback((invoker, invocation, ex) ->
AsyncRpcResult.newDefaultAsyncResult(FacadeResultUtils.returnWithFail(FacadeResultEnum.USER_FLOW_LIMIT), invocation));

咱们做了哪些额定的优化：

1）公司外部的限流控制台尚不反对热点参数限流配置，因而咱们减少了新的限流配置控制器，反对通过配置核心中动静下发限流配置。整体流程如下：

限流配置动静下发；

public class VivoCfgDataSourceConfig implements InitializingBean {
    private static final String PARAM_FLOW_RULE_PREFIX = "sentinel.param.flow.rule";
 
    @Override
    public void afterPropertiesSet() {
        // 定制配置解析对象
        VivoCfgDataSource<List<ParamFlowRule>> paramFlowRuleVivoDataSource = new VivoCfgDataSource<>(PARAM_FLOW_RULE_PREFIX, sources -> sources.stream().map(source -> JSON.parseObject(source, ParamFlowRule.class)).collect(Collectors.toList()));
        // 注册配置失效监听器
        ParamFlowRuleManager.register2Property(paramFlowRuleVivoDataSource.getProperty());
        // 初始化限流配置
        paramFlowRuleVivoDataSource.init();
 
        // 监听配置核心
        VivoConfigManager.addListener(((item, type) -> {if (item.getName().startsWith(PARAM_FLOW_RULE_PREFIX)) {paramFlowRuleVivoDataSource.updateValue(item, type);
            }
        }));
    }
}

2）原生 sentinel 指定限流热点参数的形式是两种：

第一种是指定接口办法的第 n 个参数；
第二种是办法参数继承 ParamFlowArgument，实现 ParamFlowKey 办法，该办法返回值为热点参数 value 值。

这两种形式都不是特点灵便，第一种形式不反对指定对象属性；第二种形式须要咱们革新代码，如果上线后某个接口参数没有继承 ParamFlowArgument 又想配置热点参数限流，那么只能通过改代码发版的形式解决了。因而咱们对 sentinel 组件的热点参数限流源码做了些优化，减少「指定参数对象的某个属性」作为热点参数，并且反对对象层级的嵌套。很小的代码改变，却大大不便了热点参数的配置。

革新后的热点参数校验逻辑；

public static boolean passCheck(ResourceWrapper resourceWrapper, /*@Valid*/ ParamFlowRule rule, /*@Valid*/ int count,
                                Object... args) {
 
    // 疏忽局部代码
    // Get parameter value. If value is null, then pass.
    Object value = args[paramIdx];
    if (value == null) {return true;}
 
    // Assign value with the result of paramFlowKey method
    if (value instanceof ParamFlowArgument) {value = ((ParamFlowArgument) value).paramFlowKey();}else{
        // 依据 classFieldName 指定的热点参数获取热点参数值
        if (StringUtil.isNotBlank(rule.getClassFieldName())){
            // 反射获取参数对象中的 classFieldName 属性值
            value = getParamFieldValue(value, rule.getClassFieldName());
        }
    }
    // 疏忽局部代码
}

为什么要做数据隔离？这其中有两点起因，第一点：中台存储了前台不同业务的数据，在数据查问时各业务数据不能相互影响，不能 A 业务查问到 B 业务的数据。第二点：各业务的数据量级不同、对 db 操作的压力不同，如流量隔离中咱们独自提供了一套服务集群给浏览器业务应用，那么浏览器业务应用的 db 同样须要独自配置一套，这样能力彻底和其余业务的服务压力隔离开。

vivo 评论中台应用了 MongoDB 作为存储介质（对于数据库选型及 Mongodb 利用的细节有趣味的同学能够看下咱们之前的介绍《MongoDB 在评论中台的实际》），为了隔离不同业务方的数据，评论中台提供了两种数据隔离计划：物理隔离、逻辑隔离。

不同业务方的数据存储在不同的数据库集群中，这就须要咱们零碎反对 MongoDB 的多数据源。实现过程如下：

1）寻找适合的切入点

通过剖析 spring-data-mongodb 的执行过程的源码发现，在执行所有语句前都会去做一个 getDB() 获取数据库连贯实例的动作，如下。

spring-data-mongodb db 操作源码；

private <T> T executeFindOneInternal(CollectionCallback<DBObject> collectionCallback,
        DbObjectCallback<T> objectCallback, String collectionName) {
    try {// 要害代码 getDb()
        T result = objectCallback
                .doWith(collectionCallback.doInCollection(getAndPrepareCollection(getDb(), collectionName)));
        return result;
    } catch (RuntimeException e) {throw potentiallyConvertRuntimeException(e, exceptionTranslator);
    }
}

getDB() 会执行 org.springframework.data.mongodb.MongoDbFactory 接口的 getDb() 办法，默认状况下应用 MongoDbFactory 的 SimpleMongoDbFactory 实现，看到这里咱们很天然的就能想到使用「代理模式」，用 SimpleMongoDbFactory 代理对象去替换 SimpleMongoDbFactory，并在代理对象外部为每个 MongoDB 集配创立一个 SimpleMongoDbFactory 实例。

在执行 db 操作时执行代理对象的 getDb() 操作，它只须要做两件事；

找到对应集群的 SimpleMongoDbFactory 对象
执行 SimpleMongoDbFactory.getdb() 操作。

类关系图如下。

整体的执行过程如下：

Dubbo filter 获取业务身份并设置到上下文；

private boolean setCustomerCode(Object argument) {
     // 从 string 类型参数中获取业务身份信息
    if (argument instanceof String) {if (!Pattern.matches("client.*", (String) argument)) {return false;}
        // 设置业务身份信息到上下文中
        CustomerThreadLocalUtil.setCustomerCode((String) argument);
        return true;
    } else {
        // 从 list 类型中获取参数对象
        if (argument instanceof List) {List<?> listArg = (List<?>) argument;
            if (CollectionUtils.isEmpty(listArg)) {return false;}
            argument = ((List<?>) argument).get(0);
        }
        // 从 object 对象中获取业务身份信息
        try {Method method = argument.getClass().getMethod(GET_CLIENT_CODE_METHOD);
            Object object = method.invoke(argument);
            // 校验业务身份是否非法
            ClientParamCheckService clientParamCheckService = ApplicationUtil.getBean(ClientParamCheckService.class);
            clientParamCheckService.checkClientValid(String.valueOf(object));
            // 设置业务身份信息到上下文中
            CustomerThreadLocalUtil.setCustomerCode((String) object);
            return true;
        } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {log.debug("反射获取 clientCode 失败，入参为：{}", argument.getClass().getName(), e);
            return false;
        }
    }
}

MongoDB 集群的路由代理类；

public class MultiMongoDbFactory extends SimpleMongoDbFactory {
 
    // 不同集群的数据库实例缓存：key 为 MongoDB 集群配置名，value 为对应业务的 MongoDB 集群实例
    private final Map<String, SimpleMongoDbFactory> mongoDbFactoryMap = new ConcurrentHashMap<>();
 
    // 增加创立好的 MongoDB 集群实例
    public void addDb(String dbKey, SimpleMongoDbFactory mongoDbFactory) {mongoDbFactoryMap.put(dbKey, mongoDbFactory);
    }
 
    @Override
    public DB getDb() throws DataAccessException {
        // 从上下文中获取前台业务编码
        String customerCode = CustomerThreadLocalUtil.getCustomerCode();
        // 获取该业务对应的 MongoDB 配置名
        String dbKey = VivoConfigManager.get(ConfigKeyConstants.USER_DB_KEY_PREFIX + customerCode);
        // 从连贯缓存中获取对应的 SimpleMongoDbFactory 实例
        if (dbKey != null && mongoDbFactoryMap.get(dbKey) != null) {// 执行 SimpleMongoDbFactory.getDb() 操作
            return mongoDbFactoryMap.get(dbKey).getDb();}
        return super.getDb();}
}

自定义 MongoDB 操作模板；

@Bean
public MongoTemplate createIgnoreClass() {
    // 生成 MultiMongoDbFactory 代理
    MultiMongoDbFactory multiMongoDbFactory = multiMongoDbFactory();
    if (multiMongoDbFactory == null) {return null;}
    MappingMongoConverter converter = new MappingMongoConverter(new DefaultDbRefResolver(multiMongoDbFactory), new MongoMappingContext());
    converter.setTypeMapper(new DefaultMongoTypeMapper(null));
    // 应用 multiMongoDbFactory 代理生成 MongoDB 操作模板
    return new MongoTemplate(multiMongoDbFactory, converter);
}

物理隔离是最彻底的数据隔离，然而咱们不可能为每一个业务都去搭建一套独立的 MongoDB 集群。当多个业务共用一个数据库时，就须要做数据的逻辑隔离。

逻辑隔离个别分为两种 ：

一种是表隔离：不同业务方的数据存储在同一个数据库的不同表中，不同的业务操作不同的数据表。
一种是行隔离：不同业务方的数据存储在同一个表中，表中冗余业务方编码，在读取数据时通过业务编码过滤条件来实现隔离数据目标。

从实现老本及评论业务场景思考，咱们抉择了表隔离的形式。实现过程如下：

1）初始化数据表

每次有新业务对接时，咱们都会为业务调配一个惟一的身份编码，咱们间接应用该身份编码作为业务表表名的后缀，并初始化表，例如：商城评论表 comment\_info\_vshop、社区评论表 comment\_info\_community。

2）主动寻表

间接利用 spring-data-mongodb @Document 注解反对 Spel 的能力，联合咱们的业务身份信息上下文，实现主动寻表。

主动寻表

@Document(collection = "comment_info_#{T(com.vivo.internet.comment.common.utils.CustomerThreadLocalUtil).getCustomerCode()}")
public class Comment {// 表字段疏忽}

两种隔离形式联合后的整体成果：

通过上文的这些实际，咱们很好的的撑持了不同量级的前台业务，并且做到了对业务代码无入侵，较好的解耦了技术和业务间的复杂度。另外咱们对我的项目中应用到的 Redis 集群、ES 集群对不同业务也做了隔离，大体思路和 MongoDB 的隔离相似，都是做一层代理，这里就不一一介绍了。

作者：vivo 官网商城开发团队 -Sun Daoming

关于java:vivo-评论中台的流量及数据隔离实践

一、背景

二、流量隔离

2.1 流量分组

2.2 多租户限流

三、MongoDB 数据隔离

3.1 物理隔离

3.1.2 外围代码实现

3.2 逻辑隔离

四、最初

Just My Socks（注册教程内含优惠码）

关于java:vivo-评论中台的流量及数据隔离实践

一、背景

二、流量隔离

2.1 流量分组

2.2 多租户限流

三、MongoDB 数据隔离

3.1 物理隔离

3.1.2 外围代码实现

3.2 逻辑隔离

四、最初

Just My Socks（注册教程 内含优惠码）

Just My Socks（注册教程内含优惠码）