日处理20亿数据，实时用户行为服务系统架构实践

携程实时用户行为服务做为基础服务，目前广泛应用在多个场景中，好比猜你喜欢（携程的推荐系统）、动态广告、用户画像、浏览历史等等。

以猜你喜欢为例，猜你喜欢为应用内用户提供潜在选项，提升成交效率。旅行是一项综合性的需求，用户每每须要不止一个产品。做为一站式的旅游服务平台，跨业务线的推荐，特别是实时推荐，能实际知足用户的需求，所以在上游提供打通各业务线之间的用户行为数据有很大的必要性。

携程原有的实时用户行为系统存在一些问题，包括：1）数据覆盖不全；2）数据输出没有统一格式，对众多使用方提升了接入成本；3）日志处理模块是web service，比较难支持多种数据处理策略和实现方便扩容应对流量洪峰的需求等。

而近几年旅游市场高速增加，数据量愈来愈大，而且会持续快速增加。有愈来愈多的使用需求，对系统的实时性，稳定性也提出了更高的要求。总的来讲，当前需求对系统的实时性/可用性/性能/扩展性方面都有很高的要求。
 

1、架构

 

这样的背景下，咱们按照以下结构从新设计了系统：

图1：实时用户行为系统逻辑视图

 

新的架构下，数据有两种流向，分别是处理流和输出流。

在处理流，行为日志会从客户端（App/Online/H5）上传到服务端的Collector Service。Collector Service将消息发送到分布式队列。数据处理模块由流计算框架完成，从分布式队列读出数据，处理以后把数据写入数据层，由分布式缓存和数据库集群组成。

输出流相对简单，Web Service的后台会从数据层拉取数据，并输出给调用方，有的是内部服务调用，好比推荐系统，也有的是输出到前台，好比浏览历史。系统实现采用的是Java+Kafka+Storm+Redis+MySQL+Tomcat+spring的技术栈。

1. Java：目前公司内部Java化的氛围比较浓厚，而且Java有比较成熟的大数据组件

 

2. Kafka/Storm：Kafka做为分布式消息队列已经在公司有比较成熟的应用，流计算框架Storm也已经落地，而且有比较好的运维支持环境。

 

3. Redis： Redis的HA，SortedSet和过时等特性比较好地知足了系统的需求。

 

4. MySQL: 做为基础系统，稳定性和性能也是系统的两大指标，对比NoSQL的主要选项，好比HBase和ElasticSearch，十亿数据级别上MySQL在这两方面有更好的表现，而且通过设计可以有不错的水平扩展能力。

 

目前系统天天处理20亿左右的数据量，数据从上线到可用的时间在300毫秒左右。查询服务天天服务8000万左右的请求，平均延迟在6毫秒左右。下面从实时性/可用性/性能/部署几个维度来讲明系统的设计。
 

2、实时性

 

做为一个实时系统，实时性是首要指标。线上系统面对着各类异常状况。例如以下几种状况：

1.突发流量洪峰，怎么应对；
2.出现失败数据或故障模块，如何保证失败数据重试并同时保证新数据的处理；
3.环境问题或bug致使数据积压，如何快速消解；
4.程序bug，旧数据须要从新处理，如何快速处理同时保证新数据；

 

系统从设计之初就考虑了上述状况。

首先是用storm解决了突发流量洪峰的问题。storm具备以下特性：

图2：Storm特性

 

做为一个流计算框架，和早期大数据处理的批处理框架有明显区别。批处理框架是执行完一次任务就结束运行，而流处理框架则持续运行，理论上永不中止，而且处理粒度是消息级别，所以只要系统的计算能力足够，就能保证每条消息都能第一时间被发现并处理。

对当前系统来讲，经过storm处理框架，消息能在进入kafka以后毫秒级别被处理。此外，storm具备强大的scale out能力。只要经过后台修改worker数量参数，并重启topology（storm的任务名称），能够立刻扩展计算能力，方便应对突发的流量洪峰。

对消息的处理storm支持多种数据保证策略，at least once，at most once，exactly once。对实时用户行为来讲，首先是保证数据尽量少丢失，另外要支持包括重试和降级的多种数据处理策略，并不能发挥exactly once的优点，反而会由于事务支持下降性能，因此实时用户行为系统采用的at least once的策略。这种策略下消息可能会重发，因此程序处理实现了幂等支持。

storm的发布比较简单，上传更新程序jar包并重启任务便可完成一次发布，遗憾的是没有多版本灰度发布的支持。

图3：Storm架构

 

在部分状况下数据处理须要重试，好比数据库链接超时，或者没法链接。链接超时可能立刻重试就能恢复，可是没法链接通常须要更长时间等待网络或数据库的恢复，这种状况下处理程序不能一直等待，不然会形成数据延迟。实时用户行为系统采用了双队列的设计来解决这个问题。

图4：双队列设计

 

生产者将行为纪录写入Queue1（主要保持数据新鲜），Worker从Queue1消费新鲜数据。若是发生上述异常数据，则Worker将异常数据写入Queue2（主要保持异常数据）。

这样Worker对Queue1的消费进度不会被异常数据影响，能够保持消费新鲜数据。RetryWorker会监听Queue2，消费异常数据，若是处理尚未成功，则按照必定的策略（以下图）等待或者从新将异常数据写入Queue2。

 

图5：补偿重试策略

 

另外，数据发生积压的状况下，能够调整Worker的消费游标，从最新的数据从新开始消费，保证最新数据获得处理。中间未经处理的一段数据则启动backupWorker，指定起止游标，在消费完指定区间的数据以后，backupWorker会自动中止。（以下图）

 

图6：积压数据消解

 

3、可用性

 

做为基础服务，对可用性的要求比通常的服务要高得多，由于下游依赖的服务多，一旦出现故障，有可能会引发级联反应影响大量业务。项目从设计上对如下问题作了处理，保障系统的可用性：

1.系统是否有单点？
2.DB扩容/维护/故障怎么办？
3.Redis维护/升级补丁怎么办？

4.服务万一挂了如何快速恢复？如何尽可能不影响下游应用？

 

首先是系统层面上作了全栈集群化。kafka和storm自己比较成熟地支持集群化运维；web服务支持了无状态处理而且经过负载均衡实现集群化；redis和DB方面携程已经支持主备部署，使用过程当中若是主机发生故障，备机会自动接管服务；经过全栈集群化保障系统没有单点。

另外系统在部分模块不可用时经过降级处理保障整个系统的可用性。先看看正常数据处理流程：（以下图）

 

图7：正常数据流程

 

在系统正常状态下，storm会从kafka中读取数据，分别写入到redis和mysql中。服务从redis拉取（取不到时从db补偿），输出给客户端。DB降级的状况下，数据流程也随之改变（以下图）

 

图8：系统降级-DB

 

当mysql不可用时，经过打开db降级开关，storm会正常写入redis，但再也不往mysql写入数据。数据进入reids就能够被查询服务使用，提供给客户端。另外storm会把数据写入一份到kafka的retry队列，在mysql正常服务以后，经过关闭db降级开关，storm会消费retry队列中的数据，从而把数据写入到mysql中。redis和mysql的数据在降级期间会有不一致，但系统恢复正常以后会经过retry保证数据最终的一致性。redis的降级处理也相似（以下图）

 

图9：系统降级-Redis

 

惟一有点不一样的是Redis的服务能力要远超过MySQL。因此在Redis降级时系统的吞吐能力是降低的。这时咱们会监控db压力，若是发现MySQL压力较大，会暂时中止数据的写入，下降MySQL的压力，从而保证查询服务的稳定。

为了下降故障状况下对下游的影响，查询服务经过Netflix的Hystrix组件支持了熔断模式（以下图）。

 

图10：Circuit Breaker Pattern

 

在该模式下，一旦服务失败请求在给定时间内超过一个阈值，就会打开熔断开关。在开关开启状况下，服务对后续请求直接返回失败响应，不会再让请求通过业务模块处理，从而避免服务器进一步增长压力引发雪崩，也不会由于响应时间延长拖累调用方。

开关打开以后会开始计时，timeout后会进入Half Open的状态，在该状态下会容许一个请求经过，进入业务处理模块，若是能正常返回则关闭开关，不然继续保持开关打开直到下次timeout。这样业务恢复以后就能正常服务请求。

另外，为了防止单个调用方的非法调用对服务的影响，服务也支持了多个维度限流，包括调用方AppId/ip限流和服务限流，接口限流等。
 

4、性能&扩展

 

因为在线旅游行业近几年的高速增加，携程做为行业领头羊也蓬勃发展，所以访问量和数据量也大幅提高。公司对业务的要求是能够支撑10倍容量扩展，扩展最难的部分在数据层，由于涉及到存量数据的迁移。

实时用户行为系统的数据层包括Redis和MySQL，Redis由于实现了一致性哈希，扩容时只要加机器，并对分配到新分区的数据做读补偿就能够。

MySQL方面，咱们也作了水平切分做为扩展的准备，分片数量的选择考虑为2的n次方，这样作在扩容时有明显的好处。由于携程的mysql数据库如今广泛采用的是一主一备的方式，在扩容时能够直接把备机拉平成第二台（组）主机。假设原来分了2个库，d0和d1，都放在服务器s0上，s0同时有备机s1。扩容只须要以下几步：

     1.确保s0 -> s1同步顺利，没有明显延迟
     2.s0暂时关闭读写权限
     3.确认s1已经彻底同步s0更新
     4.s1开放读写权限
     5.d1的dns由s0切换到s1

     6.s0开放读写权限

 

迁移过程利用MySQL的复制分发特性，避免了繁琐易错的人工同步过程，大大下降了迁移成本和时间。整个操做过程能够在几分钟完成，结合DB降级的功能，只有在DNS切换的几秒钟时间会产生异常。

整个过程比较简单方便，下降了运维负担，必定程度也能下降过多操做形成相似GitLab式悲剧的可能性。
 

5、部署

 

前文提到Storm部署是比较方便的，只要上传重启就能够完成部署。部署以后因为程序从新启动上下文丢失，能够经过Kafka记录的游标找到以前处理位置，恢复处理。  另外有部分状况下程序可能须要多版本运行，好比行为纪录暂时有多个版本，这种状况下咱们会新增一个backupJob，在backupJob中运行历史版本。