大数据-Storm流式框架（四）---storm容错机制

1、集群节点宕机

Nimbus服务器硬件

单点故障？可以搭建HA jStorm搭建 nimbus的HA

nimbus的信息存储到zookeeper中，只要下游没问题（进程退出）nimbus退出就不会有问题，

如果在nimbus宕机，也不能提交作业。

非Nimbus服务器硬件

supervisor故障时，该节点上所有Task任务都会超时，Nimbus会将这些Task任务重新分配到其他服务器上运行

2、进程挂掉

Worker

挂掉时，Supervisor会重新启动这个进程。如果启动过程中仍然一直失败，并且无法向zookeeper发送心跳，Nimbus会将该Worker执行的任务重新分配到其他服务器上

Supervisor

无状态（所有的状态信息都存放在Zookeeper中来管理）

快速失败（每当遇到任何异常情况，都会自动毁灭）

Nimbus

无状态（所有的状态信息都存放在Zookeeper中来管理）

快速失败（每当遇到任何异常情况，都会自动毁灭）

3、消息的完整性

从Spout中发出的Tuple，以及基于他所产生Tuple（例如上个例子当中Spout发出的句子，以及句子当中单词的tuple等）。由这些消息就构成了一棵tuple树。当这棵tuple树发送完成，并且树当中每一条消息都被正确处理，就表明spout发送消息被“完整处理”，即消息的完整性。

元组树

如果衍生元组出错，则重发根元组

根元组就是spout发送的那个元组

Acker -- 消息完整性的实现机制

Storm的拓扑当中特殊的一些任务

负责跟踪每个Spout发出的Tuple的DAG（有向无环图）

默认每个worker一个acker

保证消息至少处理一次

Storm提供了几种不同级别的保证消息处理，包括尽力而为（best effort），至少一次（at least once），以及通过Trident保证只完全处理一次（exactly once）。

此处指的是至少完全处理一次。

当元组树已经用完并且树中的每条消息都已处理完毕时，Storm会认为从水龙头发射的元组是“完全处理”的。如果在指定的超时内无法完全处理其消息树，则认为该元组失败。可以使用Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS对指定的拓扑进行配置，默认为30秒。

public interface ISpout extends Serializable {
    void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
    void close();
    void nextTuple();
    void ack(Object msgId);
    void fail(Object msgId);
}

首先，storm调用spout的nextTuple方法发射一个元组。spout使用SpoutOutputCollector在声明的流中发射元组。当发射元组的时候，spout使用messageId标记该元组。

_collector.emit(new Values("field1", "field2", 3) , msgId);

其次，当向消费闪电发送元组的时候，strom追踪该消息树。如果storm发现一个元组被完全处理了，storm就会调用spout的ack方法并将该元组的messageId传送给它。如果元组处理超时，storm就调用spout的fail方法。只会在发射该元组的spout中调用它的ack或者fail方法。

好处是：

首先，当创建了新的元组树边的时候，要通知storm。其次，当完成了一个元组的处理之后要告诉storm。通过这种方式，storm就可以知道元组被完全处理了，然后调用ack方法，或者调用fail方法，如果处理失败。

在元组树中创建一条边，称为锚点。当发送一个新元组的时候就会创建一条边。

public class SplitSentence extends BaseRichBolt {
        OutputCollector _collector;

        public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
            _collector = collector;
        }

        public void execute(Tuple tuple) {
            String sentence = tuple.getString(0);
            for(String word: sentence.split(" ")) {
                _collector.emit(tuple, new Values(word));
            }
            _collector.ack(tuple);
        }

        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("word"));
        }        
    }

每个元组通过在emit中指定输入的元组进行锚点标记。

由于被标记锚点，如果在下游处理中处理失败了，元组树顶点的元组会重新发送。

_collector.emit(new Values(word));

上述方式没有对元组进行锚点标记。当元组在下游处理失败了，根元组不会重发。输出的元组也可以标记多个锚点。如果流中有聚合或者join，就比较有用。标记了多个锚点的元组如果处理失败，就会触发多个元组树根元组重发。通过list集合为元组标记多个锚点：

List<Tuple> anchors = new ArrayList<Tuple>();
anchors.add(tuple1);
anchors.add(tuple2);
_collector.emit(anchors, new Values(1, 2, 3));

很明显，手动调用fail方法比通过元组的处理超时从而由storm调用fail方法要更快地重发根元组。由于storm使用内存来存储元组树，如果不及时的ack或者fail有可能导致内存溢出。

storm的拓扑中提供了一组acker用于追踪spout发射的每个元组及其衍生的元组，一旦发现DAG处理完了，就同创建该元组的spout进行确认。

Config.TOPOLOGY_ACKERS用于设置acker的数量。默认情况下一个worker一个acker任务。

当拓扑创建了元组，就会为其分配一个随机的64bit的id，acker使用该ID追踪spout发送的每个元组。元组树上的元组都知道这个ID。当在闪电中创建了新的元组，该ids会拷贝给新的元组。当元组确认后，元素发送消息给acker任务，以改变元组树。

当通过C衍生出D和E并且确认之后，就会从元组树中移除C。这样可以保证元组树不会过早地完成。

如果拓扑中包含多个acker，当一个元组确认后，如何知道向哪个acker发送确认消息？

storm使用hash取模的方式将一个spout的元组id跟一个acker任务绑定。

acker任务如何知道该向哪个spout任务发送确认消息？spout发射元组的时候会给合适的acker发送一个消息表示对哪个spout的元组负责。acker发现元组树完成了，就知道向哪个spout任务发送完成的消息。

acker不显式地追踪元组。如果有数十万的节点，追踪所有的元组会有耗尽acker内存的风险。acker使用一个定长的空间20字节做这个工作。这个是storm的主要创新。

acker将spout发射的元组id和一个64bit的数字（ack val）相关联。ack val代表了该元组树的状态，不管spout发射的元组及其衍生的元组有多少，它仅仅是对所有创建的元组以及确认的元组id求异或xor操作。如果最终这个值ack val变成0，表示元组树已经被完全处理。某个元组的id和该64bit的数字异或结果是0的情况极其少见，比如每秒处理10k的元组，需要5000万年才会产生一个错误，造成数据丢失。