storm保证从spout发出的每个tuple都会被完全处理。这篇文章介绍storm是怎么做到这个保证的，以及我们使用者怎么做才能充分利用storm的可靠性特点。
一个tuple被”完全处理”是什么意思？就如同蝴蝶效应一样，从spout发射的一个tuple可以引起其它成千上万个tuple因它而产生， 想想那个计算一篇文章中每个单词出现次数的topology.
1
2
3
4
5
6
7
8
9
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(1, new KestrelSpout("kestrel.backtype.com",
22133,
"sentence_queue",
new StringScheme()));
builder.setBolt(2, new SplitSentence(), 10)
.shuffleGrouping(1);
builder.setBolt(3, new WordCount(), 20)
.fieldsGrouping(2, new Fields("word"));
这个topology从一个Kestrel队列读取句子， 把每个句子分割成一个个单词， 然后发射这一个个单词： 一个源tuple（一个句子）引起后面很多tuple的产生(一个个单词）， 这个消息流大概是这样的：
统计单词出现次数的tuple树
在storm里面一个tuple被完全处理的意思是： 这个tuple以及由这个tuple所导致的所有的tuple都被成功处理。而一个tuple会被认为处理失败了如果这个消息在timeout所指定的时间内没有成功处理。 而这个timetout可以通过来指定。
如果一个消息处理成功了或者失败了会发生什么？FYI。 下面这个是spout要实现的接口:
1
2
3
4
5
6
7
8
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方法来获取下一个tuple, Spout通过open方法参数里面提供的SpoutOutputCollector来发射新tuple到它的其中一个输出消息流, 发射tuple的时候spout会提供一个message-id, 后面我们通过这个message-id来追踪这个tuple。举例来说， KestrelSpout从kestrel队列里面读取一个消息，并且把kestrel提供的消息id作为message-id, 看例子：
1
_collector.emit(new Values("field1","field2", 3),msgId);
接下来， 这个发射的tuple被传送到消息处理者bolt那里， storm会跟踪由此所产生的这课tuple树。如果storm检测到一个tuple被完全处理了， 那么storm会以最开始的那个message-id作为参数去调用消息源的ack方法；反之storm会调用spout的fail方法。值得注意的一点是， storm调用ack或者fail的task始终是产生这个tuple的那个task。所以如果一个spout被分成很多个task来执行， 消息执行的成功失败与否始终会通知最开始发出tuple的那个task。
我们再以KestrelSpout为例来看看spout需要做些什么才能保证“一个消息始终被完全处理”, 当KestrelSpout从Kestrel里面读出一条消息， 首先它“打开”这条消息， 这意味着这条消息还在kestrel队列里面， 不过这条消息会被标示成“处理中”直到ack或者fail被调用。处于“处理中“状态的消息不会被发给其他消息处理者了；并且如果这个spout“断线”了， 那么所有处于“处理中”状态的消息会被重新标示成“等待处理”.
Storm的可靠性API作为storm的使用者，有两件事情要做以更好的利用storm的可靠性特征。 首先，在你生成一个新的tuple的时候要通知storm; 其次，完成处理一个tuple之后要通知storm。 这样storm就可以检测整个tuple树有没有完成处理，并且通知源spout处理结果。storm提供了一些简洁的api来做这些事情。
由一个tuple产生一个新的tuple称为： anchoring。你发射一个新tuple的同时也就完成了一次anchoring。看下面这个例子： 这个bolt把一个包含一个句子的tuple分割成每个单词一个tuple。
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class SplitSentence implements IRichBolt {
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 cleanup() {
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}