介绍

分布式计算简单来说，是把一个大计算任务拆分成多个小计算任务分布到若干台机器上去计算，然后再进行结果汇总。 目的在于分析计算海量的数据，从雷达监测的海量历史信号中分析异常信号(外星文明)，淘宝双十一实时计算各地区的消费习惯等。

海量计算最开始的方案是提高单机计算性能，如大型机，后来由于数据的爆发式增长、单机性能却跟不上，才有分布式计算这种妥协方案。 因为计算一旦拆分，问题会变得非常复杂，像一致性、数据完整、通信、容灾、任务调度等问题也都来了。

举个例子，产品要求从数据库中100G的用户购买数据，分析出各地域的消费习惯金额等。 如果没什么时间要求，程序员小明就写个对应的业务处理服务程序，部署到服务器上，让它慢慢跑就是了，小明预计10个小时能处理完。 后面产品嫌太慢，让小明想办法加快到3个小时。
平常开发中类似的需求也很多，总结出来就是，数据量大、单机计算慢。 如果上Hadoop、storm之类成本较高、而且有点大才小用。 当然让老板买更好的服务器配置也是一种办法。

利用分片算法

小明作为一个有追求有理想的程序员，决定用介于单机计算和成熟计算框架的过度解决方案，这样成本和需求都能满足了。 分布式计算的核心在于计算任务拆分，如果数据能以水平拆分的方式，分布到5台机器上，每台机器只计算自身的1/5数据，这样即能在3小时内完成产品需求了。

如上所述，小明需要把这些数据按照一定维度进行划分。 按需求来看以用户ID划分最好，由于用户之间没有状态上的关联，所以也不需要事务性及二次迭代计算。 小明用简单的hash取模对id进行划分。

f(memberid) % 5 = ServerN

这样程序可以分别部署到5台机器上，然后程序按照配置只取对应余数的用户id，计算出结果并入库。 这种方式多机之间毫无关联，不需要进行通信，可以避免很多问题。 机器上的程序本身也不具备分布式的特性，它和单机一样，只计算自身获取到的数据即可，所以如果某台机器上程序崩溃的话，处理方式和单机一样，比如记录下处理进度，下次从当前进度继续进行后续计算。

利用消息队列

使用分片方式相对比较简单，但有如下不足之处。

• 它不具有负载均衡的能力，如果某台机器配置稍好点，它可能最先计算完，然后空闲等待着。也有可能是某些用户行为数据比较少，导致计算比较快完成。
• 还有一个弊端就是每台机器上需要手动更改对应的配置， 这样的话多台机器上的程序不是完全一样的，这样可以用远程配置动态修改的办法来解决。

小明这种方式引入了个第三方，消息队列。 小明先用一个单独的程序把用户信息推送到消息队列里去，然后各台机器分别取消费这个队列。 于是就有了3个角色：

• 推送消息的，简称Master。
• 消息队列，这里以Rabbitmq为例。
• 各个处理程序，简称Worker或Slave都行。

虽然仅仅引入了个第三方，但它已经具备了分布式计算的很多特性。

1. 计算任务分发。 Master把需要计算的用户数据，不断的推送消息队列。
2. 程序一致性。 Worker订阅相同的消息队列即可，无需更改程序代码。
3. 任意扩容。 由于程序完全一样，意味着如果想要加快速度，重复部署一份程序到新机器即可。 当然这是理论上的，实际当中会受限于消息队列、数据库存储等。
4. 容灾性。 如果5台中某一台程序挂了也不影响，利用Rabbitmq的消息确认机制，机器崩溃时正在计算的那一条数据会在超时，在其他节点上进行消费处理。

Hadoop简介

Hadoop介绍已经相当多了，这里简述下比如:"Hadoop是一套海量数据计算存储的基础平台架构"，分析下这句话。

• 其中计算指的是MapReduce，这是做分布式计算用的。
• 存储指的是HDFS，基于此上层的有HBase、Hive，用来做数据存储用的。
• 平台，指可以给多个用户使用，比如小明有一计算需求，他只需要按照对应的接口编写业务逻辑即可，然后把程序以包的形式发布到平台上，平台进行分配调度计算等。 而上面小明的分布式计算设计只能给自己使用，如果另外有小华要使用就需要重新写一份，然后单独部署，申请机器等。Hadoop最大的优势之一就在于提供了一套这样的完整解决方案。

下面找了介绍Hadoop的概览图，跟小明的设计做对比下:

• 图中“大数据计算任务” 对应小明的100G用户数据的计算任务。
• ”任务划分“ 对应Master和消息队列。
• “子任务” 对应Worker的业务逻辑。
• ”结果合并“ 对应把每个worker的计算结果入库。
• “计算结果” 对应入库的用户消费习惯数据。

PS：为了方便描述，把小明设计的分布式计算，叫做小和尚。

MapReduce

由于MapReduce计算输入和输出都是基于HDFS文件，所以大多数公司的做法是把mysql或sqlserver的数据导入到HDFS，计算完后再导出到常规的数据库中，这是MapReduce不够灵活的地方之一。 MapReduce优势在于提供了比较简单的分布式计算编程模型，使开发此类程序变得非常简单，像之前的MPI编程就相当复杂。

狭隘的来讲，MapReduce是把计算任务给规范化了，它可以等同于小和尚中Worker的业务逻辑部分。 MapReduce把业务逻辑给拆分成2个大部分，Map和Reduce，可以先在Map部分把任务计算一半后，扔给Reduce部分继续后面的计算。 当然在Map部分把计算任务全做完也是可以的。 关于Mapreduce实现细节部分不多解释，有兴趣的同学可以查相关资料或看下楼主之前的C#模拟实现的博客【探索C#之微型MapReduce】。

如果把小明产品经理的需求放到Hadoop来做，其处理流程大致如下：

1. 把100G数据导入到HDFS
2. 按照Mapreduce的接口编写处理逻辑，分Map、Reduce两部分。
3. 把程序包提交到Mapreduce平台上，存储在HDFS里。
4. 平台中有个叫Jobtracker进程的角色进行分发任务。 这个类似小和尚的Master负载调度管理。
5. 如果有5台机器进行计算的话，就会提前运行5个叫TaskTracker的slave进程。 这类似小和尚worker的分离版，平台把程序和业务逻辑进行分离了， 简单来说就是在机器上运行个独立进程，它能动态加载、执行jar或dll的业务逻辑代码。
6. Jobtracker把任务分发到TaskTracker后，TaskTracker把开始动态加载jar包，创建个独立进程执行Map部分，然后把结果写入到HDFS上。
7. 如果有Reduce部分，TaskTracker会创建个独立进程把Map输出的HDFS文件，通过RPC方式远程拉取到本地，拉取成功后，Reduce开始计算后续任务。
8. Reduce再把结果写入到HDFS中
9. 从HDFS中把结果导出。

这样一看好像是把简单的计算任务给复杂化了，其实如果只有几台计算任务的话，使用Mapreduce确实是杀鸡用牛刀了。 如果有TB、PB级别的数据、跑在成百上千台计算节点上，Mapreduce的优势才会体现出来。 其计算框架图架构如下：

离线计算

通常称Mapreduce及小和尚这种计算为离线计算，因为它对已经持久化的文件数据进行计算，不能实时响应。 还有个原因就是它的处理速度比较慢，它的输入和输出源都是基于HDFS设计，如果数据不是一开始就写入到HDFS上，就会涉及到数据导入导出，这部分相对耗费时间。 而且它的数据流动是基于文件系统的，Map部分输出的数据不是直接传送到Reduce部分，而是先写入HDFS再进行传送。

处理速度慢也是Mapreduce的不足之处，促使了后面实时计算的诞生。
另外个缺点是Mapreduce的计算任务流比较单一，它只有Map、Reduce两部分。 简单的可以只写一部分逻辑来解决，如果想拆分成多个部分，如逻辑A、逻辑B、逻辑C等， 而且一部分计算逻辑依赖上一次计算结果的话，MapReduce处理起来就比较困难了。 像storm框架解决此类问题的方案，也称为流式计算，下一章继续补充。

实时计算

接上篇，离线计算是对已经入库的数据进行计算，在查询时对批量数据进行检索、磁盘读取展示。 而实时计算是在数据产生时就对其进行计算，然后实时展示结果，一般是秒级。 举个例子来说，如果有个大型网站，要实时统计用户的搜索内容，这样就能计算出热点新闻及突发事件了。 按照以前离线计算的做法是不能满足的，需要使用到实时计算。

小明作为有理想、有追求的程序员开始设计其解决方案了，主要分三部分。

• 每当搜索内容的数据产生时，先把数据收集到消息队列，由于其数据量较大，以使用kafka为例。 这个收集过程是一直持续的，数据不断产生然后不断流入到kafka中。
• 要有一个能持续计算的框架，一旦收集到数据，计算系统能实时收到数据，根据业务逻辑开始计算，然后不断产生需要的结果，这里以storm为例。
• 根据结果进行实时展示并入库， 可以一边展示一边入库，对外提供实时查询的服务。这里的入库可以是基于内存的Redis、MongoDB，也可是基于磁盘的HBase、Mysql、SqlServer等。

其流程图如下:

storm简介

通常都介绍Storm是一个分布式的、高容错的实时计算系统。 “分布式”是把数据分布到多台上进行计算，“高容错”下面谈，这里主要细节介绍下“实时计算”的实现。

storm有个角色叫topology，它类似mapreduce的job，是一个完整的业务计算任务抽象。 上章谈到hadoop的缺点在于数据源单一依赖HDFS，storm中Spout角色的出现解决了这个问题。 在Spout内部我们可以读取任意数据源的数据，比如Redis、消息队列、数据库等等。 而且spout可以是多个，这样更好的分类，比如可以SpoutA读取kafka，SpoutB读取Redis。 示例如下：

public class CalcPriceSpout : BaseRichSpout
{private SpoutCollector Collector;public override void NexData(){//读取各种数据源，Redis、消息队列、数据库等Collector.emit("消息")}
}

代码中NexData是storm的核心方法，它一直被storm循环调用着， 在方法里我们实时读取kafka的消息，然后把消息通过Collector组件发射到各个计算节点里，它类似小和尚中的Master。 这样应用每产生一条数据，会实时收集到kafka，然后被NextData消费，发射到节点开始计算。 NextData读取的消息时在内存中，然后直接通过网络流动到节点机器上的内存中开始计算，不会持久化到磁盘上。

因为速度比较快，所以叫实时计算，也有叫持续计算，意思是可以非常快的一直进行计算，至于叫什么都可以。

流式计算

主流的流式计算有S4、StreamBase、Borealis，其storm也具有流式计算的特性。 流式计算是指“数据能像液体水一样不断的在各个节点间流动，每个节点都可以对“数据(液体水)”进行计算，然后产生新的数据，继续像水一样流动”。如图：

图中Spout就是水龙头，它不断的通过NextData产生数据，然后流动各个Bolt中。 Bolt是各个计算节点上的计算逻辑，它拿到数据后开始计算，完成后流向另外一个，直到完成。 其Bolt也可以是任意个，这比Mapreduce只能分成Map、Reduce两部分好多了。 这样可以在BlotA中计算中间值，然后通过这个中间值去任意数据源拉取数据后，在流动到下一步处理逻辑中， 这个中间值直接在内存中，通过网络流动BlotB上。 其大大增加了其适用范围和灵活度，Spout和bolt的数据流动构成了一个有向无环图。 Bolt示例代码如下。

public class CalcProductPriceBolt : BaseRichBolt
{private BoltCollector Collector;public override void Execute(Tuple<string, string> input){//Result=计算计算计算。//Collector.Emit("Reulst"); 流动到另外一个节点
    }
}

数据流动图：

归纳总结

结合上篇，发现Hadoop离线计算的计算要求是把业务逻辑包上传到平台上，数据导入到HDFS上，这样才能进行计算。 其产生的结果数据是展示之前就计算好的，另外它的计算是按批次来的，比如很多公司的报表，都是每天凌晨开始计算前一天的数据，以便于展示。 其数据是不动的，计算逻辑也是不动的。

Storm的流式计算同样是把计算逻辑包上传到平台上，由平台调度，计算逻辑是不动的。 但数据可以是任意来源的，不断在计算节点进行流动。 也即是说在数据产生的时刻，就开始进行流动计算，它展示的结果数据是实时变化的。 其数据是流动的，计算逻辑是不动的。storm把产生的每条数据当成一个消息来处理，其内部也是通过消息队列组件zeromq来完成的。

高容错性

storm提供了各级别的可靠性保证，一消息从Spout流动到boltA，在流动boltB， 那storm会通过唯一值不断异或的设计去监测这个消息的完成情况，这个监测是一个和业务逻辑类似的bolt，不过它是有storm自身实现的，叫Acker，它的任务就是接收各个消息任务的完成状态，然后告诉Spout这个消息是否已经完全处理。下面是几种异常处理情况：

• BoltB所在的节点挂了或消息异常，那么这条消息就没有处理完，Spout可在超时后重新发射该数据即可。
• Acker所在节点挂了后，即当前节点监控的消息完全情况，会全部丢失，Spout会在消息超时做后续处理。
• 如果Spout所在节点挂了，那Spout发射的数据也会全部丢失， 这时可在消息队列中设置超时时间，如果没有一直没对消息进行Ack的话，那么这条消息会重新让其他的Spout重新接收到。这部分需要单独在消息队列中配置，另外storm消息的Ack确认对性能有一定影响，可根据消息的重要性是否要开启它。
• 如果storm平台级别的组件挂了，平台会尝试重启失败的组件，storm除nimbus组件外都是多节点点部署，挂了某一节点，不会对任务计算有所影响。

下篇写消息保证机制及改造小和尚的设计。

作者：蘑菇先生 出处： http://mushroom.cnblogs.com/