总体认识storm包括概念，场景，组成

本帖最后由 nettman 于 2014-6-9 12:40 编辑
阅读本文章可以带着下面问题：
1.storm是如何产生的？

2.为什么storm会比其他解决方案受欢迎？

3.storm有几个进程，各自的作用是什么？

4.Storm记录级容错的基本原理？

5.storm内核使用什么编写的？

6.storm被淘宝用来做什么？

淘宝场景分析：

伴随着信息科技日新月异的发展，信息呈现出爆发式的膨胀，人们获取信息的途径也更加多样、更加便捷，同时对于信息的时效性要求也越来越高。举个搜索场景中的例子，当一个卖家发布了一条宝贝信息时，他希望的当然是这个宝贝马上就可以被卖家搜索出来、点击、购买啦，相反，如果这个宝贝要等到第二天或者更久才可以被搜出来，估计这个大哥就要骂娘了。再举一个推荐的例子，如果用户昨天在淘宝上买了一双袜子，今天想买一副泳镜去游泳，但是却发现系统在不遗余力地给他推荐袜子、鞋子，根本对他今天寻找泳镜的行为视而不见，估计这哥们心里就会想推荐你妹呀。其实稍微了解点背景知识的码农们都知道，这是因为后台系统做的是每天一次的全量处理，而且大多是在夜深人静之时做的，那么你今天白天做的事情当然要明天才能反映出来啦。
迫切需要设计一个实时系统

根据场景分析，我们迫切需要一个大数据实时系统，这个实时系统该如何设计，需要解决什么问题

低延迟：都说了是实时计算系统了，延迟是一定要低的。
高性能：性能不高就是浪费机器，浪费机器是要受批评的哦。
分布式：系统都是为应用场景而生的，如果你的应用场景、你的数据和计算单机就能搞定，那么不用考虑这些复杂的问题了。我们所说的是单机搞不定的情况。
可扩展：伴随着业务的发展，我们的数据量、计算量可能会越来越大，所以希望这个系统是可扩展的。
容错：这是分布式系统中通用问题。一个节点挂了不能影响我的应用。

那么如果能够解决上面问题，可以说有不少解决方案。而且各有千秋，随便举一种方案，使用消息队列+分布在各个机器上的工作进程就ok啦。
但是我们在是使用过程中如何保证质量，而且质量往往决定了受欢迎的程度。

1.如何容易在上面开发应用程序。

亲，你设计的系统需要应用程序开发人员考虑各个处理组件的分布、消息的传递吗？如果是，那有点麻烦啊，开发人员可能会用不好，也不会想去用。

2.消息不丢失。

用户发布的一个宝贝消息不能在实时处理的时候给丢了，对吧？更严格一点，如果是一个精确数据统计的应用，那么它处理的消息要不多不少才行。这个要求有点高哦。

3.消息严格有序。

有些消息之间是有强相关性的，比如同一个宝贝的更新和删除操作消息，如果处理时搞乱顺序完全是不一样的效果了。

面对上面的的问题，和受欢迎程度-------storm横空出世

我们来看一下storm的适用场景。

流数据处理：Storm可以用来处理源源不断流进来的消息，处理之后将结果写入到某个存储中去。
分布式rpc：由于storm的处理组件是分布式的，而且处理延迟极低，所以可以作为一个通用的分布式rpc框架来使用。当然，其实我们的搜索引擎本身也是一个分布式rpc系统。

什么是Storm详细可以查看storm入门介绍

Storm是一个开源的实时计算系统，它提供了一系列的基本元素用于进行计算：Topology、Stream、Spout、Bolt等等。

在Storm中，一个实时应用的计算任务被打包作为Topology发布，这同Hadoop的MapReduce任务相似。但是有一点不同的是：在Hadoop中，MapReduce任务最终会执行完成后结束；而在Storm中，Topology任务一旦提交后永远不会结束，除非你显示去停止任务。

计算任务Topology是由不同的Spouts和Bolts，通过数据流（Stream）连接起来的图。下面是一个Topology的结构示意图：

其中包含有：

Spout：Storm中的消息源，用于为Topology生产消息（数据），一般是从外部数据源（如Message Queue、RDBMS、NoSQL、Realtime Log）不间断地读取数据并发送给Topology消息（tuple元组）。

Bolt：Storm中的消息处理者，用于为Topology进行消息的处理，Bolt可以执行过滤，聚合，查询数据库等操作，而且可以一级一级的进行处理。

最终，Topology会被提交到storm集群中运行；也可以通过命令停止Topology的运行，将Topology占用的计算资源归还给Storm集群。

Storm数据流模型

数据流（Stream）是Storm中对数据进行的抽象，它是时间上无界的tuple元组序列。在Topology中，Spout是Stream的源头，负责为Topology从特定数据源发射Stream；Bolt可以接收任意多个Stream作为输入，然后进行数据的加工处理过程，如果需要，Bolt还可以发射出新的Stream给下级Bolt进行处理。

下面是一个Topology内部Spout和Bolt之间的数据流关系：

Topology中每一个计算组件（Spout和Bolt）都有一个并行执行度，在创建Topology时可以进行指定，Storm会在集群内分配对应并行度个数的线程来同时执行这一组件。

那么，有一个问题：既然对于一个Spout或Bolt，都会有多个task线程来运行，那么如何在两个组件（Spout和Bolt）之间发送tuple元组呢？

Storm提供了若干种数据流分发（Stream Grouping）策略用来解决这一问题。在Topology定义时，需要为每个Bolt指定接收什么样的Stream作为其输入（注：Spout并不需要接收Stream，只会发射Stream）。

目前Storm中提供了以下7种Stream Grouping策略：Shuffle Grouping、Fields Grouping、All Grouping、Global Grouping、Non Grouping、Direct Grouping、Local or shuffle grouping，具体策略可以参考这里。

首先我们通过一个 storm 和hadoop的对比来了解storm中的基本概念。

首先我们通过一个 storm 和hadoop的对比来了解storm中的基本概念。

	Hadoop	Storm
系统角色	JobTracker	Nimbus
	TaskTracker	Supervisor
	Child	Worker
应用名称	Job	Topology
组件接口	Mapper/Reducer	Spout/Bolt

接下来我们再来具体看一下这些概念。

Nimbus：负责资源分配和任务调度。
Supervisor：负责接受nimbus分配的任务，启动和停止属于自己管理的worker进程。
Worker：运行具体处理组件逻辑的进程。
Task：worker中每一个spout/bolt的线程称为一个task. 在storm0.8之后，task不再与物理线程对应，同一个spout/bolt的task可能会共享一个物理线程，该线程称为executor。

下面这个图描述了以上几个角色之间的关系。

图3-1

Topology：storm中运行的一个实时应用程序，因为各个组件间的消息流动形成逻辑上的一个拓扑结构。
Spout：在一个topology中产生源数据流的组件。通常情况下spout会从外部数据源中读取数据，然后转换为topology内部的源数据。Spout是一个主动的角色，其接口中有个nextTuple()函数，storm框架会不停地调用此函数，用户只要在其中生成源数据即可。
Bolt：在一个topology中接受数据然后执行处理的组件。Bolt可以执行过滤、函数操作、合并、写数据库等任何操作。Bolt是一个被动的角色，其接口中有个execute(Tuple input)函数,在接受到消息后会调用此函数，用户可以在其中执行自己想要的操作。
Tuple：一次消息传递的基本单元。本来应该是一个key-value的map，但是由于各个组件间传递的tuple的字段名称已经事先定义好，所以tuple中只要按序填入各个value就行了，所以就是一个value list.
Stream：源源不断传递的tuple就组成了stream。

10. stream grouping：即消息的partition方法。Storm中提供若干种实用的grouping方式，包括shuffle, fields hash, all, global, none, direct和localOrShuffle等

相比于s4, puma等其他实时计算系统，storm最大的亮点在于其记录级容错和能够保证消息精确处理的事务功能。下面就重点来看一下这两个亮点的实现原理。

Storm记录级容错的基本原理

首先来看一下什么叫做记录级容错？storm允许用户在spout中发射一个新的源tuple时为其指定一个message id, 这个message id可以是任意的object对象。多个源tuple可以共用一个message id，表示这多个源 tuple对用户来说是同一个消息单元。storm中记录级容错的意思是说，storm会告知用户每一个消息单元是否在指定时间内被完全处理了。那什么叫做完全处理呢，就是该message id绑定的源tuple及由该源tuple后续生成的tuple经过了topology中每一个应该到达的bolt的处理。举个例子。在图4-1中，在spout由message 1绑定的tuple1和tuple2经过了bolt1和bolt2的处理生成两个新的tuple，并最终都流向了bolt3。当这个过程完成处理完时，称message 1被完全处理了。

图4-1

在storm的topology中有一个系统级组件，叫做acker。这个acker的任务就是追踪从spout中流出来的每一个message id绑定的若干tuple的处理路径，如果在用户设置的最大超时时间内这些tuple没有被完全处理，那么acker就会告知spout该消息处理失败了，相反则会告知spout该消息处理成功了。在刚才的描述中，我们提到了”记录tuple的处理路径”，如果曾经尝试过这么做的同学可以仔细地思考一下这件事的复杂程度。但是storm中却是使用了一种非常巧妙的方法做到了。在说明这个方法之前，我们来复习一个数学定理。

A xor A = 0.

A xor B…xor B xor A = 0，其中每一个操作数出现且仅出现两次。

storm中使用的巧妙方法就是基于这个定理。具体过程是这样的：在spout中系统会为用户指定的message id生成一个对应的64位整数，作为一个root id。root id会传递给acker及后续的bolt作为该消息单元的唯一标识。同时无论是spout还是bolt每次新生成一个tuple的时候，都会赋予该tuple一个64位的整数的id。Spout发射完某个message id对应的源tuple之后，会告知acker自己发射的root id及生成的那些源tuple的id。而bolt呢，每次接受到一个输入tuple处理完之后，也会告知acker自己处理的输入tuple的id及新生成的那些tuple的id。Acker只需要对这些id做一个简单的异或运算，就能判断出该root id对应的消息单元是否处理完成了。下面通过一个图示来说明这个过程。

图4-1 spout中绑定message 1生成了两个源tuple，id分别是0010和1011.

图4-2 bolt1处理tuple 0010时生成了一个新的tuple，id为0110.

图4-3 bolt2处理tuple 1011时生成了一个新的tuple，id为0111.

图4-4 bolt3中接收到tuple 0110和tuple 0111，没有生成新的tuple.

可能有些细心的同学会发现，容错过程存在一个可能出错的地方，那就是，如果生成的tuple id并不是完全各异的，acker可能会在消息单元完全处理完成之前就错误的计算为0。这个错误在理论上的确是存在的，但是在实际中其概率是极低极低的，完全可以忽略。

Storm在淘宝

由于storm的内核是clojure编写的(不过大部分的拓展工作都是java编写的)，为我们理解它的实现带来了一定的困难，好在大部分情况下storm都比较稳定，当然我们也在尽力熟悉clojure的世界。我们在使用storm时通常都是选择java语言开发应用程序。

在淘宝，storm被广泛用来进行实时日志处理，出现在实时统计、实时风控、实时推荐等场景中。一般来说，我们从类kafka的metaQ或者基于hbase的timetunnel中读取实时日志消息，经过一系列处理，最终将处理结果写入到一个分布式存储中，提供给应用程序访问。我们每天的实时消息量从几百万到几十亿不等，数据总量达到TB级。对于我们来说，storm往往会配合分布式存储服务一起使用。在我们正在进行的个性化搜索实时分析项目中，就使用了timetunnel + hbase + storm + ups的架构，每天处理几十亿的用户日志信息，从用户行为发生到完成分析延迟在秒级。