如何为信息流内容中心设计一个高效的处理链路，简单聊聊QQ看点在这方面的演进过程

Workflow Engine诞生的背景

内容处理主要包括过滤、打标和内容本身的处理三部分，如下图所示:

提几个问题，这几个问题都有一定的共性：

1. 往内容处理系统新增一个工程模块，怎么做比较快速？
2. 抖音视频内容处理中心（全球每天共产生100亿的小视频）,如何快速的调整针对某个国家或地区的内容处理策略，以便能符合当地的监管要求？
3. 华为无线oss系统管理着自己多个制式的网络通信设备（包括基站、控制器等）、并可以对其进行远端软件升级。不同制式的设备，升级前准备工作、步骤、命令、参数、通信协议差异巨大，这个系统如何设计才能应付灵活的扩展？

可以说，对于稍微复杂一些的处理系统，包括互联网产品、电信产品，都需要一个Workflow Engine（为行文方便，以下简称WE）来提高系统整体的可靠性、扩展性。QQ看点从成立之初，采用简单的DAG（Directed Acyclic Graph）模式来实现内容处理，随着业务不断发展壮大，逐渐过渡到了WE处理模式。

WE应该具备的功能

1. 从字面意思解读，WE负责工作流程的编排，对整个工作流负责，不能因为某个子系统出现问题而导致整个工作流受阻。流的速度既不能过高也不能过低，过高则压垮处理子系统，过低则影响处理效率。
2. 能够组装出多种工作流，以便满足产品、运营的各种策略，并能快速调整。

3. 抽象出通用的非业务逻辑，减少新子系统开发量，快速迭代上线。

• 通过HBase的状态位字段，所有模块形成一个有向无环图（DAG），并依据产品策略做不同处理逻辑。待本模块的所有前置模块都处理完毕后，才开始能执行当前模块。

• 模块之间无直接通信，只和存储HBase打交道。

• 优点：实现简单，能够快速完成产品功能的迭代上线。

• 缺点：处理效率低，对存储HBase依赖大，产品策略调整不灵活，无完整的监控调用链，故障定位麻烦。

3、针对DAG优化的备选方案

• 通过Kafka的Topic订阅、发布，所有模块形成一个有向无环图，并依据产品策略做不同处理逻辑。本模块订阅所有依赖模块的Topic，待所有Topic都收到后再开始处理，处理完毕后发布本模块的Topic。
• 模块之间无直接通信，只和消息中间件Kafka打交道。
• 优点：改进了依赖存储HBase处理效率低的问题。
• 缺点：当模块之间的依赖关系变得复杂后，Topic的管理会非常麻烦，仍然存在产品策略调整不灵活、无完整监控调用链、故障定位麻烦的问题。当一个模块依赖2个以上（含）的Topic，实现层面的挑战也非常大。

• WE有三部分组成，Spout、Dispatch、管理Portal，Spout负责源源不断的提供待处理文章；Dispatch负责和具体业务模块打交道；管理Portal用于配置和调整工作流。通过L5来实现服务的注册、发现、容灾与备份。
• Spout多台机器之间通过Zookeeper和一致性Hash，构成容灾和备份系统。
• Dispatch多机之间通过分布式锁互斥处理文章，并通过Portal管理系统配置的工作流来和具体业务模块打交道。为方便处理结果回溯，工作流执行情况上报到TDW保存。
• 优化效果：文章处理耗时从以前的30分钟缩短到3分钟之内，大大提高了文章处理效率，模块接入也方便很多。

5、工作流中间状态的考量

7、往中台方向推进

WE与业务模块的业务协议

message FieldInfo{    optional bytes bytes_key = 1;//hbase中的字段名    optional bytes bytes_value = 2;};
message ReqBody{    optional bytes bytes_rowkey = 1;    repeated FieldInfo rpt_msg_field_info = 2;// 模块输入参数字段    //  下面字段主要用于：调度模块将当前请求的超时时间传递给具体业务模块，业务模块根据    // 这个信息可以做一些优先级处理。如果从接收到请求到过了uint64_time_out还没有回包，    // 调度模块则认为超时。    optional uint64 uint64_time_out = 3;//当前请求的超时时间，例如10000毫秒，精确到毫秒,};
enum RetEnum{    SUCC = 0x00;//处理成功，指某个步骤成功处理完成，和文章本身是否合法、能否推送给看点没有关系。
    FAIL_PARSE_REQ_PKG = 0X40;// 解析请求包失败(调用方也不可能收到这样的错误，纯属定义全集场景)
    REQ_PKG_PARAM_INVALID = 0X41;// 请求参数非法，终止本篇文章的处理+告警上报+记录此文章rowkey
    REQ_MUST_PARAM_MISS = 0X42;// 必选或者前置参数缺失，终止本篇文章的处理+告警上报+记录此文章rowkey
    SVR_INNER_ERR = 0x43;//服务内部错误，流式中心重试N次后：终止本篇文章的处理+告警上报+记录此文章rowkey
    FAIL_BACK_SVR = 0X44;//后端服务错误，流式中心重试N次后：终止本篇文章的处理+告警上报+记录此文章rowkey
    FAIL_PACK_RSP_PKG = 0X45;//回包组装失败(调用方也不可能收到这样的错误，纯属定义全集场景)
    SVR_OVER_LOAD = 0X81;//本服务过载，流式中心重新分配业务的机器，如果都是服务过载则当前文章sleep一段时间+报警。
    NO_AUTH = 0X90;//没有权限
    OVER_LIMITS = 0X91;//调用超频
    FAIL_BACK_TIMEOUT = 0X92;//后端模块超时
    FATAL_ERROR = 0X93;//后端模块致命错误};
message RspBody{    optional RetEnum enum_ret = 1;    optional bytes bytes_msg = 2;    optional bytes bytes_rowkey = 3;    repeated FieldInfo rpt_msg_field_info = 4;//模块输出参数字段};

Tcp同步等待的方式，简化了业务模块的开发模式：接受输入参数->处理->返回结果。

以每天1000万篇文章计算，QPS峰值在350/s左右（假设峰值为均值的3倍，下同）。若后面文章量变成100亿，QPS峰值将达到35万/s，Dispatch和业务模块可以横向扩展，架构无需变动。Spout模块，因扫描的是全量未处理文章，需要在前面做一次业务分片，然后在分片里面通过一致性Hash来实现容灾和备份处理。

总结