Storm的消息保证机制

Storm提供了三种不同层次的消息保证机制，分别是At Most Once,At Least Once,Exactly Once.消息保证机制依赖于消息是否被完全处理
消息完全处理
每个从Spout（Storm中数据源点）发出的Tuple（Storm中最小的消息单元）可能会生成成千上万个新的Tuple,形成一颗Tuple树，当整颗Tuple树的节点都被成功处理了，我们就说从Spout发出的Tuple被完全处理了。我们可以通过下面的例子来更好的诠释消息是否被完全处理这个概念：

            TopologyBuilder builder = new TopologyBuilder();
            builder.setSpout("sentences",new KafkaSpout(spoutConfig),spoutNum);
            builder.setBolt("split",new SplitSentence(),10).shuffleGrouping("sentences");
            builder.setBolt("count",new WordCount(),20).fieldsGrouping("split",new Fields("word"));

这个Topology从Kafka(一个开源的分布式消息队列)读取信息发往下游，下游的Bolt将收到句子分割成单独的单词，并进行计数。每一个从Spout发送出来的Tuple会衍生出多个新的Tuple,从Spout发送出来的Tuple以及后续衍生出来的Tuple形成一颗Tuple树，下图是一颗Tuple树示例：

Paste_Image.png

上图中所有的Tuple都被成功处理了，我们才认为Spout发出的Tuple被完全处理。如果在一个固定的时间内(这个时间可以配置，默认为30秒)，有至少一个Tuple处理失败或者超时，则认为整颗Tuple树处理失败，即从Spout发出的Tuple处理失败

如何实现不同层次的消息保证机制
Storm提供的三种不同消息保证机制中，利用Spout,Bolt以及Acker的组合可以实现At Most Once以及At Least Once语义，Storm在At Least Once的基础上进行了一次封装（Trident），从而实现Exactly Once语义
Storm的消息保证机制中，如果需要实现At Most Once语义，只需要满足下面任何一条即可：

        1.关闭ACK机制，即Acker数目设置为0
        2.Spout不实现可靠性传输
                Spout发送消息是使用不带message ID的API
                不实现fail函数
        3.Bolt不把处理成功或失败的消息发送给Acker

如果需要实现At Least Once 语义，则需要同时保证如下几条：

         1.开启ACK机制，即Acker数目大于0
         2.Spout实现可靠性传输保证
                  Spout发送消息附带message 的ID
                  如果收到Acker的处理失败反馈，需要进行消息重传，即实现fail函数
         3.Bolt在处理成功或失败后需要调用相应的方法通知Acker

实现Exactly Once语义，则需要在At Least Once的基础上进行状态的存储，用来防止重复发送的数据被重复处理，在Storm中使用Trident API实现

下图中，每种消息保证机制中左边的字母表示上游发送的消息，右边的字母表示下游收到的消息。从图中直到，At Most Once中，消息可能会丢失(上游发送了两个A，下游只收到一个A)；At Least Once中，消息不会丢失，可能会重复(上游只发送了一个B，下游收到两个B)；Exactly Once中，消息不丢失，不重复，因此需要在At Least Once的基础上保存相应的状态，表示上游的哪些消息已经发送到下游 ，防止同一条消息发送多次给下游的情况

Paste_Image.png

Tuple的完全处理需要Spout、Bolt以及Acker（Storm中用来记录某颗Tuple树是否被完全处理的节点）协调完成，如图所示。从Spout发送Tuple到下游，并把相应信息通知给Acker，整颗Tuple树中某个Tuple被成功处理了都会通知Acker,待整颗Tuple树都被完全处理完成之后，Acker将成功处理信息返回给Spout；如果某个Tuple处理失败，或者超时，Acker将会给Spout发送一个处理失败的消息，Spout根据Acker的返回信息以及用户对消息保证机制的选择判断是否需要进行消息重传

Paste_Image.png

不同消息可靠性保证的使用场景
对于Storm提供的三种消息可靠性保证，优缺点以及使用场景如下所示：

  可靠性保证层次        优点        缺点        使用场景
   At Most Once      处理速度快  数据可能丢失  对处理速度要求高，且对数据丢失容忍度高的场景
   At Least Once    数据不会丢失  数据可能重复   不能容忍数据丢失，可以容忍数据重复的场景
   Exactly  Once    数据不会丢失不会重复  处理速度慢  对数据不丢不重性质要求高非常高，且处理速度要求没那么高，比如支付金额

不同层次的可靠性如何实现
如何实现可靠的Spout

    实现可靠的Spout需要在nextTuple函数中发送消息时，调用带msgID的emit方法，然后实现失败消息重传(fail函数)，参考如下所示：
      //想实现可靠的Spout,需要实现如下两点
      //1.在nextTuple函数中调用emit函数时需要带一个msgId,用来表示当前的消息(如果消息发送失败会用msgID作为参数回调fail函数)
     //2.自己实现fail函数，进行重发(注意，在storm中没有msgID和消息的对应关系，需要自己维护)

  public void nextTuple(){
            collector.emit(new Values(curNum+" ",round+" "),curNum+":"+round);
    }
  @Override
    public void fail(Object msgId){
              String tmp = (String)msgId；
              String[] args = tmp.split(":");
              collector.emit(new Values(args[0],args[1]),msgId);
      }

如何实现可靠的Bolt
Storm提供了两种不同类型的Bolt,分别是BaseRichBolt和BaseBasicBolt，都可以实现可靠性消息传递，不过BaseRichBolt需要自己做很多周边的事情(建立anchor树，以及手动ACL/FAIL,通知Acker),使用场景更广泛，而BaseBasicBolt则由Storm帮忙实现很多周边的事情，实现起来方便简单，但是使用场景单一。如果使用这两个Bolt实现(不)可靠的消息传递如下所示：

      //BaseRichBolt实现不可靠消息传递
      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(new Values(word));
                              }
                              _collector.ack(tuple);
                      }

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

      //BaseRichBolt实现可靠的Bolt
      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"));
                      }
          }

    下面的示例会可以建立Multi-anchoring
          List<Tuple> anchors = new ArrayList<Tuple>();
          anchors.add(tuple1);
          anchors.add(tuple2);
          _collector.emit(anchors,new Values(1,2,3));
            
          public class SplitSentence extends BaseBasicBolt{
                        public void execute(Tuple tuple,BasicOutputCollector collector){ 
                             String sentence = tuple.getString(0);
                              for(String word:sentence.split(" ")){
                                          collector.emit(new Values(word));
                              }
                          }

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