审核中台业务数据进审升级之路

原创仇智慧搜狐技术产品

本文字数：3850字

预计阅读时间：15 分钟

1.背景

1.1 相关名词介绍

1.1.1 业务端

业务端是指某个产品线下的具体的一个业务。比如视频产品线细分业务端：举报视频、56视频、搜狐视频、千帆视频；用户产品线细分业务端：用户签名、用户头像、用户头图、用户昵称。评论产品线细分业务端：弹幕抓取、弹幕直播、弹幕加一、弹幕普通、弹幕举报、弹幕热剧、视频评论、头条评。等等目前接入的业务端共计有58个。相同的产品线划分不同的业务端，是为了应用不同的审核策略，比如审核的时效不同，审核的优先级不同，审核标准的严格或宽松的程度不同。

1.1.2 审核中台

业务端生成的业务数据发送给审核中台。审核中台前置有机器审核，机器审核通过后再经过人工审核。审核结束后，审核中台将审核结果发送给业务端。业务端根据审核结果对业务数据做出相应的处理：数据露出或者数据删除。审核中台是为了把控用户生成数据的安全性与合法性。

1.1.3 领域模型

领域模型（Domain Model），是完成从需求分析到面向对象设计的一座桥梁，领域模型是指对需求所涉及的领域的建模，所以也叫业务对象模型，是描述业务用例实现的对象模型。它是对业务角色和业务实体之间应该如何联系和协作以执行业务的一种抽象。下文中提到的业务领域模型是对业务端数据进审的业务流程进行建模的模型。模型名称为：MBusiness，关键属性为inTopic（进审消息topic），inGroup（进审消息消费者group），outTopic（出审消息topic），outGroup（出审消息生产者group），type（业务类型）。（inTopic、outTopic：是每个业务端对应一组。）

1.2 审核中台介绍

众所周知，互联网平台中都存在审核流程，以实现净化网络环境的目的，而在搜狐视频平台中也不例外。搜狐视频审核中台就是这样一个承接了搜狐矩阵大部分的业务审核需求的平台。

审核中台数年来对接了58个业务端，由于各产品线下的新业务频出，每出现一个新的业务端，新的业务端生成的业务数据都需要发送到审核中台以某种审核标准进行审核操作。

1.3 业务痛点介绍

接入每一个新的业务端，虽然审核标准不同，但进审的流程都是相似的。共有两大痛点：

痛点1

业务端的数据发送给审核中台有的通过http协议，有的通过mq协议。通讯协议不统一，数据规范也不统一。

痛点2

每接入新的业务端都需要重复的编写相似的进审流程代码，占用了大量的开发成本，并且代码重复率太高，大大的降低了代码的质量，增加了系统的维护成本，对开发和系统运维不友好，不利于系统的健康发展。

2. 规范化改造

2.1 规范通讯协议

由于业务线众多，为实现业务解耦审核中台采用消息中间件的方式对接业务线，本文中使用的是RocketMQ，RocketMQ是一个高可用、高性能、高可靠的分布式消息队列，相对于kafka更适合处理业务系统之间的消息。

2.2 规范处理流程

1）在审核中台中预设了三种类型的标准数据进审流程，包括：视频、专辑、图文。（预设的标准流程如不满足新业务需求，可定制开发新的流程，该三种标注流程目前已经涵盖所有的产品线和业务端。）

2）在审核中台定义了业务领域模型，MBusiness，该领域模型关键属性为inTopic（进审消息topic），inGroup（进审消息消费者group），outTopic（出审消息topic），outGroup（出审消息生产者group），type（业务类型）。

3）定义了数据的流转规范：

a、inTopic为进审topic，在inTopic下业务端作为Producer生产消息，审核中台作为Consumer消费消息。

b、outTopic为出审topic，在outTopic下审核中台作为Producer生产消息，业务端作为Consumer消费消息。

4）当有新业务需要接入时，首先创建业务领域模型，MBusiness，设置好相应的属性，存入数据库中。

5）业务端首先作为消息生产者，向inTopic发送进审消息；然后业务端作为消息消费者，监听outTopic的出审消息。

6）审核中台设置了业务接入检测器，不断扫描是否有新的业务领域模型创建。

7）当扫描到MBusiness时，自动创建进审消息消费者Consumer，拉取inTopic下的消息，根据MBusiness的type按预设的标准数据进审流程（视频、专辑、图文）进行数据进审处理。

8）自动创建出审消息生产者Producer，等待审核结束后负责将审核结果发送到outTopic下。

9）业务端通过监听outTopic的出审消息，获取审核结果。

3. 自动化改造

3.1 业务接入检测器

实现业务端的数据自动化进审，是依靠业务接入检测器装置完成的。业务接入检测器启动步骤如下：

1、系统启动后，业务接入检测器立即启动。

2、业务接入检测器持有2个map，consumerMap：key为inTopic，value为具体的消费者程序。producerMap：key为业务名，value为具体的生产者程序，初始状态，2个map全为空。

3、业务接入检测器每3分钟读取数据库业务领域模型表所有数据MBusiness。

4、读取MBusiness的inTopic，如果consumerMap中没有，说明是新业务，则以该inTopic和inGroup创建消费者，以outTopic和outGroup创建生产者，设置type对应的预设数据进审流程等待数据进审处理，并将inTopic与消费者存入consumerMap中，业务名与生产者存入producerMap中。

5、等待审核结束后，通过业务名从producerMap中获取对应的生产者，发送审核结果。

6、系统停止时，遍历consumerMap和producerMap获取所有的消费者和生产者，进行生产者和消费者的关停操作。

3.2 数据自动化流转

下面介绍一下进审数据如何自动化流转的：

1、业务端作为生产者生产进审消息，发送到消息队列inTopic下。

2、审核中台作为消费者，监听消息队列inTopic下的消息变化。

3、每监听到inTopic下的新消息，则按预设的数据进审流程进行数据进审处理。

4、等待审核结束后，通过业务名从producerMap中获取对应的生产者，并发送审核结果到outTopic下。

5、业务端作为消费者监听outTopic下的消息变化，获取审核结果。

3.3 源码示例

3.3.1自动进审源码示例

下述源码是对业务端自动进审的描述，其中，VideoConsumerCallback、ContentConsumerCallback、PlayListConsumerCallback、BroadlistConsumerCallback是预设的视频、专辑、图文的标准处理流程，如源码所示，bean ConsumerManager创建时，通过init方法从数据库加载已接入审核中台的业务端（businessDao.listValid()），使用InGroup和InTopic自动构建消费者，根据业务的类型按预设的标准处理流程进行处理。并通过flush方法创建了一个定时任务，每隔5分钟检测是否有新的业务端接入，若有则自动构建消费者，根据业务的类型按预设的标准处理流程进行处理。实现了DisposableBean在程序销毁时，通过destroy方法，对消费者bean进行销毁。

@Slf4j
@Component
public class ConsumerManager implements ApplicationRunner, DisposableBean {
    private Map<String, Consumer> consumerMap = Collections.synchronizedMap(new HashMap<>());
    @Resource
    private VideoConsumerCallback videoConsumerCallback;
    @Resource
    private ContentConsumerCallback contentConsumerCallback;
    @Resource
    private PlayListConsumerCallback playListConsumerCallback;
    @Resource
    private BroadlistConsumerCallback broadlistConsumerCallback;
    @Resource
    private MBusinessDao businessDao;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        init();
        flush();
    }

    private void flush() {
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {

            @Override
            public void run() {
                init();
            }

        }, 1000 * 60 * 5L, 1000 * 60 * 5L);
    }

    private void init() {
        List<MBusiness> businessList = businessDao.listValid();
        if (CollectionUtils.isEmpty(businessList)) {
            return;
        }

        for (MBusiness business : businessList) {
            if (consumerMap.containsKey(business.getInTopic())) {
                continue;
            }

            Consumer consumer = initConsumer(business);
            if (consumer == null) {
                continue;
            }

            consumer.start();
            consumerMap.put(business.getInTopic(), consumer);
            log.info("topic:{} group:{} start consumer.", business.getInTopic(), business.getInGroup());
        }

    }

    @Override
    public void destroy() throws Exception {
        if (consumerMap.isEmpty()) {
            return;
        }
        for (Consumer consumer : consumerMap.values()) {
            consumer.shutdown();
        }
    }

    private Consumer initConsumer(MBusiness business) {
        if (StringUtils.isBlank(business.getInGroup()) || StringUtils.isBlank(business.getInTopic())) {
            return null;
        }

        Consumer consumer = new Consumer(business.getInGroup(), business.getInTopic());
        if (EnterType.VIDEO.getType().equals(business.getEnterType())) {
            consumer.setConsumerCallback(videoConsumerCallback);
        }
        if (EnterType.CONTENT.getType().equals(business.getEnterType())) {
            consumer.setConsumerCallback(contentConsumerCallback);
        }
        if (EnterType.PLAYLIST.getType().equals(business.getEnterType())) {
            consumer.setConsumerCallback(playListConsumerCallback);
        }
        if (EnterType.BROADLIST.getType().equals(business.getEnterType())) {
            consumer.setConsumerCallback(broadlistConsumerCallback);
        }

        consumer.setPullThresholdForQueue(1);
        return consumer;
    }

    public Set<String> listTopic() {
        return consumerMap.keySet();
    }

    public Consumer getConsumer(String topic) {
        return consumerMap.get(topic);
    }
}

3.3.2自动出审源码示例

下述源码是对业务端自动出审的描述，其中，bean ProducerManager创建时，通过init方法从数据库加载已接入审核中台的业务端（businessDao.listValid()），使用OutGroup和OutTopic自动构建生产者，并通过flush方法创建了一个定时任务，每隔3分钟检测是否有新的业务端接入，若有则自动构建生产者，用于审核结束时通过消息将审核结果下发到业务端。

@Slf4j
@Component
public class ProducerManager implements ApplicationRunner, DisposableBean {
    private Map<String, Producer> producerMap = Collections.synchronizedMap(new HashMap<>());
    private Map<String, Producer> producerTypeMap = Collections.synchronizedMap(new HashMap<>());
    private Map<String, String> webhookMap = Collections.synchronizedMap(new HashMap<>());

    @Resource
    private MBusinessDao businessDao;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        init();
        flush();
    }

    private void flush() {
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {

            @Override
            public void run() {
                init();
            }

        }, 1000 * 10 * 3L, 1000 * 60 * 3L);
    }

    private void init() {
        List<MBusiness> businessList = businessDao.listValid();
        if (CollectionUtils.isEmpty(businessList)) {
            return;
        }

        for (MBusiness business : businessList) {
            if (StringUtils.isNotBlank(business.getOutWebhook()) && !webhookMap.containsKey(business.getType())) {
                webhookMap.put(business.getType(), business.getOutWebhook());
            }

            if (producerMap.containsKey(business.getOutTopic())) {
                if (!producerTypeMap.containsKey(business.getType())) {
                    producerTypeMap.put(business.getType(), producerMap.get(business.getOutTopic()));
                    log.info("business type:{} topic:{} group:{} init producer.", business.getType(), business.getOutTopic(), business.getOutGroup());
                }

                continue;
            }

            Producer producer = initProducer(business);
            if (producer == null) {
                continue;
            }

            producer.start();
            producerMap.put(business.getOutTopic(), producer);
            producerTypeMap.put(business.getType(), producer);
            log.info("business type:{} topic:{} group:{} init producer.", business.getType(), business.getOutTopic(), business.getOutGroup());
        }
    }

    @Override
    public void destroy() throws Exception {
        if (producerMap.isEmpty()) {
            return;
        }
        for (Producer producer : producerMap.values()) {
            producer.shutdown();
        }
    }

    public Producer initProducer(MBusiness business) {
        if (StringUtils.isBlank(business.getOutGroup()) || StringUtils.isBlank(business.getOutTopic())) {
            return null;
        }

        return new Producer(business.getOutGroup(), business.getOutTopic());
    }

    public String getWebhook(String type) {
        if (StringUtils.isBlank(type)) {
            return null;
        }
        return webhookMap.get(type);
    }

    public Producer getProducer(String type) {
        if (StringUtils.isBlank(type)) {
            return null;
        }
        return producerTypeMap.get(type);
    }

}

4 .总结

通过规范通讯协议，规范进审数据格式、规范数据进审流程，将各个产品线下的多种多样的业务端抽象为视频、专辑、图文三种类型，分别预设了标准化的数据进审流程；并设计了业务接入检测器装置，自动化的检测新业务接入并自动处理业务数据的进审，避免了新的业务端接入时的重复开发工作，极大的提高了开发效率。

继续滑动看下一个