一、背景

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

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

二、流量隔离

2.1 流量分组

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并不受该路由标签的影响。

2.2 多租户限流

大流量的业务咱们通过独自的集群隔离进来了。然而独立部署集群老本高，不能为每个前台业务都独立部署一套集群。大部分状况下多个业务还是须要共用一套集群的，那么共用集群的服务遇到了突发流量如何解决呢？没错，限流呗！然而目前很多限流都是一刀切的形式对接口整体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());
        }
    }
    // 疏忽局部代码
}

三、MongoDB数据隔离

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

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

3.1 物理隔离

不同业务方的数据存储在不同的数据库集群中，这就须要咱们零碎反对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( )操作。

类关系图如下。

整体的执行过程如下：

3.1.2 外围代码实现

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);
}

3.2 逻辑隔离

物理隔离是最彻底的数据隔离，然而咱们不可能为每一个业务都去搭建一套独立的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 逻辑隔离

四、最初

评论

发表回复取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

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

一、背景

二、流量隔离

2.1 流量分组

2.2 多租户限流

三、MongoDB数据隔离

3.1 物理隔离

3.1.2 外围代码实现

3.2 逻辑隔离

四、最初

评论

发表回复 取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

发表回复取消回复