[TOC] # ack是什么 通过Ack机制，spout发送出去的每一条消息，都可以确定是被成功处理或失败处理， 从而可以让开发者采取动作。比如在Meta中，成功被处理，即可更新偏移量，当失败时，重复发送数据。 因此，通过Ack机制，很容易做到保证所有数据均被处理，一条都不漏。 **另外需要注意的，当spout触发fail动作时，不会自动重发失败的tuple，需要spout自己重新获取数据，手动重新再发送一次** ack机制即， spout发送的每一条消息， * 在规定的时间内，spout收到Acker的ack响应，即认为该tuple 被后续bolt成功处理 * 在规定的时间内，没有收到Acker的ack响应tuple，就触发fail动作，即认为该tuple处理失败， * 或者收到Acker发送的fail响应tuple，也认为失败，触发fail动作 **另外Ack机制还常用于限流作用： 为了避免spout发送数据太快，而bolt处理太慢，常常设置pending数，当spout有等于或超过pending数的tuple没有收到ack或fail响应时，跳过执行nextTuple， 从而限制spout发送数据。** 通过`conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, pending);`设置spout pend数。 # 如何使用Ack机制 spout 在发送数据的时候带上msgid 设置acker数至少大于0；`Config.setNumAckers(conf, ackerParal);` 在bolt中完成处理tuple时，执行`OutputCollector.ack(tuple)`, 当失败处理时，执行`OutputCollector.fail(tuple)`; 推荐使用IBasicBolt， 因为IBasicBolt 自动封装了`OutputCollector.ack(tuple)`, 处理失败时，请抛出FailedException，则自动执行`OutputCollector.fail(tuple)` # 如何关闭Ack机制 有2种途径 spout发送数据是不带上msgid 设置acker数等于0 # 基本实现 Storm 系统中有一组叫做"acker"的特殊的任务，它们负责跟踪DAG（有向无环图）中的每个消息。 acker任务保存了spout id到一对值的映射。第一个值就是spout的任务id，通过这个id，acker就知道消息处理完成时该通知哪个spout任务。第二个值是一个64bit的数字，我们称之为"ack val"， 它是树中所有消息的随机id的异或计算结果。 ~~~ <TaskId,<RootId,ackValue>> Spoutid,<系统生成的id,ackValue> Task-0,64bit,0 ~~~ ack val表示了整棵树的的状态，无论这棵树多大，只需要这个固定大小的数字就可以跟踪整棵树。当消息被创建和被应答的时候都会有相同的消息id发送过来做异或。 每当acker发现一棵树的ack val值为0的时候，它就知道这棵树已经被完全处理了    # 分析 storm的消息容错机制 数据在处理中出现异常时,需要保证消息被完整处理 Spout ---A---B---C---D 期望:当中一个环节出现异常时,Spout能够重新发送一份数据 问题: Spout如何知道一条消息的处理状态 ~~~ 成功:ack(Object msgid) 失败:fail(Object msgid) ~~~ Bolt如何告知Spout消息处理的状态 ~~~ collector.emit(new Value()) collector.ack() //当消息处理成功时 collector.fail() //当消息处理失败时 ~~~ # ACK机制 spout发送一条数据出去,需要知道数据处理成功和失败的状态,如果失败进行消息的重新发送 1. 自定义spout实现BaseRichSpout,覆写ack,fail方法 2. 自定义的spout发送数据的时候,需要指定messageId,messageId是一个object.如果messageId为null,Storm不会跟踪元组,也不会收到回调. 3. 当消息处理成功或失败之后,storm框架会将messageId传回来.如果消息要重发,直接通过messageId找到或直接转化成数据内容进行重发 4. 自定义Bolt实现BaseRichBolt 5. 在Bolt的execute中进行两个操作 5.1 发送数据的时候,需要指定血缘关系,锚点 `collector.emit(父tuple, new 子Tuple);` 5.2 当execute处理完业务逻辑的时候,需要告诉storm框架当前阶段的处理状态 `collector.ack(tuple);` 如果在编写storm程序,在bolt环节忘了手动ack或者fail,怎么办? storm会等待反馈,达到超时阈值之后,就直接给fail. 如果在编写storm程序时,在bolt环节忘了标识锚点,怎么办? 忘了标识锚点,及时完了标识血缘关系.storm会认为你不关系后面阶段的处理状况 Storm BaseRichBolt API过于繁琐,就开了另外一个api:BaseBasicBolt 如果实现了BaseBasicBolt就不需要锚点,不需要手动ack或fail