数据源、storm应用、结果集，三者之间的关系

Storm是一个分布式的、可靠的、容错的数据流处理系统。使用storm应用场景可以稳定支撑着实时统计、对账、监控、数据同步、风控等多个业务。其中订单实时统计是需要实时跟进的，他对于数据准确性还有性能等各方面都有较高要求。我们可以通过对订单实时统计来描述使用storm时，所遇到的各个方面的问题，比如：可靠性、准确性以及它的性能等方面。

第一版：流程走通

在没有使用storm的时候，显示实时统计数据一般有两种方案：①在数据库里执行count、sum等聚合查询，这种是属于简单快速的实现方案，但缺点就是容易出现慢查询的情况；②在业务代码里对统计指标做累加，可以满足指标的快速查询，但统计逻辑耦合到业务代码，维护不方便，而且错误数据定位和修正不方便。所以如果想要解决指标快速查询或者解耦业务和统计，那基于storm的实时计算方案可以满足这两点需求。

一个storm应用的基本结构有三部分：数据源、storm应用、结果集。storm应用从数据源读取数据，经过计算后，把结果持久化或发送消息给其他应用。在数据源方面，最早尝试在业务代码里打日志的方式，但总有业务分支无法覆盖，采集的数据不全。我们的业务数据库是mysql，随后尝试基于mysql binlog的数据源，采用了阿里开源的canal，可以做到完整的收集业务数据变更。

在结果数据的处理上，我们把统计结果持久化到了mysql，并通过另一个后台应用的RESTful API对外提供服务，一个mysql就可以满足数据的读写需求。

但是为了提升实时统计应用吞吐量，需要提升消息的并发度。spout里设置了消息缓冲区，只要消息缓冲区不满，就会源源不断从消息源canal拉取数据，并把分发到多个bolt处理。

其中对于实时计算有个常见问题，就是通过订单实时统计的案例，可以抽象出一些基于storm实时计算的共性问题。

第二版：性能提升

为了确保数据的准确性，会把所有的统计指标持久化放在一个数据库事务里。一笔订单状态更新后，会在一个事务里有两类操作：

订单的历史状态也在数据库里存着，要与历史状态对比决定统计逻辑，并把最新的状态持久化。storm的应用本身是无状态的，需要使用存储设备记录状态信息。当大家知道实时计算好用后，各产品都希望有实时数据，统计逻辑越来越复杂。店铺、商品、用户等多个指标的写操作都是在一个事务里commit，这一简单粗暴的方式早期很好满足的统计需求，但是对于update操作持有锁时间过长，严重影响了并发能力。

为此做了数据库事务的瘦身（去除历史状态的mysql持久化，是通过单条binlog消息的前后状态对比，这样就做到了统计逻辑上的无状态。但又如何保证消息有且只有处理一次，为此引入了一个redis用于保存最近24小时内已成功处理的消息binlog偏移量，而storm的消息分发机制又可以保证相同消息总是能分配到一个bolt，避免线程安全问题。）

统计业务拆分，先是线上业务和公司内部业务分离，随后又把线上业务按不同产品拆分。这个不仅仅是bolt级别的拆分，而是在spout就完全分开。随着统计应用拆分，在canal和storm应用之间加上消息队列。canal不支持多消费者，而实时统计业务也不用关系数据库底层迁移、主从切换等维护工作，加上消息队列能把底层数据的维护和性能优化交给更专业的团队。

热点数据在mysql里做了分桶。比如，通常一个店铺天级别的统计指标在mysql里是一行数据。如果这个店铺有突发的大量订单，会出现多个bolt同时去update这行数据，出现数据热点，mysql里该行数据的锁竞争异常激烈。我们把这样的热点数据做了分桶，实验证明在特定场景下可以有一个数量级吞吐量提升。

所以第二版的主要变化在于引入了MQ，并使用redis作为消息状态的存储。而且由最初的一个应用，被拆成了多个应用。

第三版：准确性提升

经过第二版的优化，实时统计的吞吐量已经不成问题，但还是遇到了做大数据最重要的准确性的问题。

统计口径是会变化的，同样是GMV，一年前和现在的算法可能有变化。而实时统计只能按照当时的算法来做计算。有可能出现一段时间周期内的GMV，前一段是按旧算法来计算，后一段按新算法来计算，提供的数据就不准确了。所以实时统计难免会出现bug，有不准确的结果，修复错误数据是个难题。

为了解决这个问题，凡是涉及到两天以前数据的，一律由离线计算提供，最终展示给用户的数据，就是历史离线统计数据，并上今日昨日实时统计数据。因为离线统计有数据准备、建模、统计的过程，要花费几个小时，每天的凌晨很可能还得不到前一天的离线统计结果。一旦统计口径有变化，只需要重跑离线统计任务就可修复历史数据，做到了冷热数据分离。

消息状态管理：storm不提供消息状态管理，而且为了达到水平扩展，最好是消息之间无状态。对于大数据量、低精度的应用，需要做到无状态。而像订单实时统计这样数据量不算太大，但精度要求极高的场景，需要记录消息处理状态。而为了应付重启、分布式扩展的场景，往往需要额外的介质来存储状态。状态信息往往是kv形式的读写，我们在实际的应用中，使用过redis、HBase作为存储。

对于准确性要求高的场景，需要保证数据正确的只消费一次。storm的有三种消息处理模式：①若不实现ack和fail方法，无论后续处理结果如何，消息只会发送一次，必定不能满足高准确性；②若实现了ack和fail方法，只有调用了ack方法才会任务处理成功，否则会重试。可能会出现消息重复，在并发场景下重复又意味着可能出现乱序；③trident每个micro batch作为整体只成功处理一次，但也是无法保证消息真的只正确的处理一次，比如数据已经处理完毕并持久化，但向数据源ack时失败，就可能会有重试。

对于消息重复、乱序的场景，不是简单的消息幂等能解决，有以下的处理思路：①使用前面提到的状态管理的办法，识别出重复、乱序的数据；②业务逻辑中，兼容重复、乱序数据，比如维护一个业务状态机，把异常数据剔除。

对于时序判断，尽量不用使用时间戳，因为在分布式系统里，各服务器时间不一致是很常见的问题。

我们会尝试在运行过程中重启消息源、storm应用、存储/MQ等下游系统，或者制造网络丢包、延迟等异常，手工触发可能的消息丢失、重复、乱序场景，来验证我们的应用能否对应这些异常情况。

对于需要消息可靠处理的场景，是不适合复杂拓扑的，部分失败如何回滚，是否要全部bolt处理完毕才ack，是需要面对的问题。过长的拓扑链路，里面的慢速逻辑会拖慢整体性能。这时可以考虑使用更简化的拓扑，不同的逻辑之间尽量解耦，需要使用bolt的结果时，可以把数据持久化或者推送到MQ。

在生产环境中最必不可少的就是监控，除了服务器的基础监控，还加了不少storm特有的监控。

①消息延迟：消息在业务系统的时间戳与storm应用的当前时间戳对比，大于一定阈值则告警，不同应用的阈值会不同；

②消息处理时长、fail数：这两个都可以由storm的接口获取，数值偏大很可能是出了问题；

③应用TPS：记录应用的emit、ack、fail数的变化趋势，帮助分析应用的运行情况；

④任务级监控：每台服务器的worker、executor数量，这也可以通过storm接口获取。

除此之外，会有各类应用特有的监控，一般都是离线计算的结果与实时计算结果对比。对于数据同步类的应用，数据量比较大，可能会使用采样的方式做校验。