目前闲鱼的实际生产部署环境越来越复杂，对各种服务的横向依赖交织在一起，对运行环境的纵向依赖也越来越复杂。 　　

　　

　　 　　

　　

　　当服务出现问题时，能否在海量数据中及时定位问题根源，成为考验闲鱼服务能力的严峻挑战。 　　

　　

　　线上出现问题，往往需要十几分钟，甚至更长时间才能找到问题原因。因此，能够快速自动诊断问题的系统应运而生，而快速诊断的基础是高性能的实时数据处理系统。 　　

　　

　　该实时数据处理系统需要以下能力： 　　

　　

　　实时数据采集，实时分析，复杂计算，分析结果持久化。可以处理各种数据。包括应用程序日志、主机性能监控指标和调用链接图。可靠性高。系统没有问题，数据不能丢失。高性能、低延迟。数据处理延迟小于3秒，支持每秒数千万次数据处理。本文不涉及问题自动诊断的具体分析模型，只讨论整个实时数据处理环节的设计。 　　

　　

　　输入输出定义 　　

　　

　　为了理解系统的运行，我们定义了系统的整体输入和输出。 　　

　　

　　输入 　　

　　

　　服务日志(包括traceid、时间戳、客户端IP、服务器IP、耗时、返回代码、服务名和方法名)。 　　

　　

　　环境监测数据(指示器名称、IP、时间戳、指示器值)。比如CPU、JVM GC时间、JVM GC时间消耗、数据库指标。 　　

　　

　　输出 　　

　　

　　一段时间内服务错误的根本原因，每个服务的错误分析结果用有向无环图表示。(根节点是被分析的错误节点，叶节点是错误根节点。叶子可以是外部依赖的服务错误或JVM异常等。). 　　

　　

　　架构设计 　　

　　

　　在实际的系统运行过程中，随着时间的推移，不断产生日志数据和监控数据。 　　

　　

　　每个生成的数据都有自己的时间戳。而实时传输这些带时间戳的数据，就像水在不同的管道里流动一样。 　　

　　

　　 　　

　　

　　如果将连续实时数据与自来水进行比较，数据处理过程类似于自来水生产过程： 　　

　　

　　 　　

　　

　　自然，我们也把实时数据的处理过程分解成几个阶段：采集、传输、预处理、计算、存储、计算、持久化。 　　

　　

　　的整体系统架构设计如下： 　　

　　

　　 　　

　　

　　采集 　　

　　

　　采用阿里自主研发的SLS日志服务产品(包括Logtail LogHub组件)，Logtail为采集客户端。 　　

　　

　　选择Logtail是因为其性能优异、可靠性高、插件扩展机制灵活。闲鱼可以自定义自己的采集插件，实现各种数据的实时采集。 　　

　　

　　传输 　　

　　

　　Log可以理解为一个数据发布和订阅组件，类似于Kafka的功能。作为数据传输通道，更稳定，更安全。 　　

　　

　　详细的对比文章参考： 　　

　　

　　https://yq.aliyun.com/articles/35979?SPM=5176.1069562 . 1996646101 . search click result . 6f 2c 7 FBE 6g 3 xgp预处理 　　

　　

　　实时数据预处理部分采用blink流计算处理组件(开源版本称为blink，Blink是基于Blink的Ali内部增强版)。 　　

　　

　　目前常用的实时流计算开源产品有Jstorm、Spark Stream和Flink: 　　

　　

　　由于Jstorm没有中间计算状态，其计算过程中所需的中间结果必须依赖外部存储，这将导致频繁的IO，影响其性能。Spark Stream本质上是用微小的批处理来模拟实时计算，但实际上还是有一定的延迟。Flink出色的状态管理机制保证了其计算的性能和实时性，并提供了完整的SQL表达式，使得流计算变得更加容易。计算与持久化 　　

　　

　　经过数据预处理，最终生成呼叫链路聚合日志和主机监控数据，主机监控数据将独立存储在TSDB时间序列数据库中，供后续统计分析使用。 　　

/p> 　　

TSDB 由于其针对时间指标数据的特别存储结构设计，非常适合做时序数据的存储与查询。

　　

调用链路日志聚合数据，提供给 Cep/Graph Service 做诊断模型分析。

　　

Cep/Graph Service 是闲鱼自研的一个应用，实现模型分析、复杂的数据处理以及外部服务进行交互，同时借助 RDB 实现图数据的实时聚合。

　　

最后 Cep/Graph Service 分析的结果作为一个图数据，实时转储在 Lindorm 中提供在线查询。Lindorm 可以看作是增强版的 Hbase，在系统中充当持久化存储的角色。

　　

详细设计与性能优化

　　

采集

　　

日志和指标数据采集使用 Logtail，整个数据采集过程如图：

　　

　　

　　

　　

其提供了非常灵活的插件机制，共有四种类型的插件：

　　

Inputs：输入插件，获取数据。Processors：处理插件，对得到的数据进行处理。Aggregators：聚合插件，对数据进行聚合。Flushers：输出插件，将数据输出到指定 Sink。由于指标数据(比如 CPU、内存、JVM 指标)的获取需要调用本地机器上的服务接口获取，因此应尽量减少请求次数，在 Logtail 中，一个 Input 占用一个 Goroutine。

　　

闲鱼通过定制 Input 插件和 Processors 插件，将多个指标数据(比如 CPU、内存、JVM 指标)在一个 Input 插件中通过一次服务请求获取(指标获取接口由基础监控团队提供)。

　　

并将其格式化成一个 Json 数组对象，在 Processors 插件中再拆分成多条数据，以减少系统的 IO 次数同时提升性能。

　　

传输

　　

数据传输使用 LogHub，Logtail 写入数据后直接由 Blink 消费其中的数据，只需设置合理的分区数量即可。

　　

分区数要大于等于 Blink 读取任务的并发数，避免 Blink 中的任务空转。

　　

预处理

　　

预处理主要采用 Blink 实现，主要的设计和优化点：

　　

①编写高效的计算流程

　　

Blink 是一个有状态的流计算框架，非常适合做实时聚合、Join 等操作。在我们的应用中只需要关注出现错误的的请求上相关服务链路的调用情况。

　　

因此整个日志处理流分成两个流：

　　

服务的请求入口日志作为一个单独的流来处理，筛选出请求出错的数据。其他中间链路的调用日志作为另一个独立的流来处理，通过和上面的流 Join On Traceid 实现出错服务依赖的请求数据筛选。

　　

　　

　　

如上图所示通过双流 Join 后，输出的就是所有发生请求错误相关链路的完整数据。

　　

②设置合理的 State 生命周期

　　

Blink 在做 Join 的时候本质上是通过 State 缓存中间数据状态，然后做数据的匹配。

　　

而如果 State 的生命周期太长会导致数据膨胀影响性能，如果 State 的生命周期太短就会无法正常关联出部分延迟到来的数据，所以需要合理的配置 State 生存周期，对于该应用允许最大数据延迟为 1 分钟。

　　

使用niagara作为statebackend，以及设定state数据生命周期，单位毫秒 state.backend.type=niagara state.backend.niagara.ttl.ms=60000 ③开启 MicroBatch/MiniBatch

　　

MicroBatch 和 MiniBatch 都是微批处理，只是微批的触发机制上略有不同。原理上都是缓存一定的数据后再触发处理，以减少对 State 的访问从而显著提升吞吐，以及减少输出数据量。

　　

开启join blink.miniBatch.join.enabled=true 使用 microbatch 时需要保留以下两个 minibatch 配置 blink.miniBatch.allowLatencyMs=5000 防止OOM，每个批次最多缓存多少条数据 blink.miniBatch.size=20000 ④Dynamic-Rebalance 替代 Rebalance

　　

Blink 任务在运行时最忌讳的就是存在计算热点，为保证数据均匀使用 Dynamic Rebalance，它可以根据当前各 Subpartition 中堆积的 Buffer 的数量，选择负载较轻的 Subpartition 进行写入，从而实现动态的负载均衡。

　　

相比于静态的 Rebalance 策略，在下游各任务计算能力不均衡时，可以使各任务相对负载更加均衡，从而提高整个作业的性能。

　　

开启动态负载 task.dynamic.rebalance.enabled=true ⑤自定义输出插件

　　

数据关联后需要将统一请求链路上的数据作为一个数据包通知下游图分析节点，传统的方式是通过消息服务来投递数据。

　　

但是通过消息服务有两个缺点：

　　

其吞吐量和 RDB 这种内存数据库相比还是较大差距(大概差一个数量级)。在接受端还需要根据 traceid 做数据关联。我们通过自定义插件的方式将数据通过异步的方式写入 RDB，同时设定数据过期时间。

　　

在 RDB 中以

　　

写入的同时只将 traceid 做为消息内容通过 MetaQ 通知下游计算服务，极大的减少了 MetaQ 的数据传输压力。

　　

图聚合计算

　　

Cep/Graph 计算服务节点在接收到 MetaQ 的通知后，综合根据请求的链路数据以及依赖的环境监控数据，会实时生成诊断结果。

　　

诊断结果简化为如下形式：

　　

　　

　　

　　

说明本次请求是由于下游 JVM 的线程池满导致的，但是一次调用并不能说明该服务是不可用的根本原因，需要分析整体的错误情况，那就需要对图数据做实时聚合。

　　

聚合设计如下(为了说明基本思路，做了简化处理)：

　　

首先利用 Redis 的 Zrank 能力为根据服务名或 IP 信息为每个节点分配一个全局唯一排序序号。为图中的每个节点生成对应图节点编码，编码格式。对于头节点：头节点序号|归整时间戳|节点编码。对于普通节点：|归整时间戳|节点编码。由于每个节点在一个时间周期内都有唯一的 Key，因此可以将节点编码作为 Key 利用 Redis 为每个节点做计数。同时消除了并发读写的问题。利用 Redis 中的 Set 集合可以很方便的叠加图的边。记录根节点，即可通过遍历还原聚合后的图结构。聚合后的结果大致如下：

　　

　　

　　

　　

这样最终生成了服务不可用的整体原因，并且通过叶子节点的计数可以实现根因的排序。

　　

收益

　　

系统上线后，整个实时处理数据链路的延迟不超过 3 秒。闲鱼服务端问题的定位时间从十多分钟甚至更长时间下降到 5 秒内。大大的提升了问题定位的效率。

　　

展望

　　

目前的系统可以支持闲鱼每秒千万的数据处理能力。后续自动定位问题的服务可能会推广到阿里内部更多的业务场景，随之而来的是数据量的成倍增加，因此对于效率和成本提出了更好的要求。

　　

未来我们可能做的改进：

　　

能够自动的减少或者压缩处理的数据。复杂的模型分析计算也可以在 Blink 中完成，减少 IO，提升性能。支持多租户的数据隔离。