一、简介

storm可以确保spout发送出来的每个消息都会被完整的处理。本章将会描述storm体系是如何达到这个目标的，并将会详述开发者应该如何使用storm的这些机制来实现数据的可靠处理。

二、理解消息被完整处理

一个消息(tuple)从spout发送出来，可能会导致成百上千的消息基于此消息被创建。

我们来思考一下流式的“单词统计”的例子：

storm任务从数据源（Kestrel queue）每次读取一个完整的英文句子；将这个句子分解为独立的单词，***，实时的输出每个单词以及它出现过的次数。

本例中，每个从spout发送出来的消息（每个英文句子）都会触发很多的消息被创建，那些从句子中分隔出来的单词就是被创建出来的新消息。

这些消息构成一个树状结构，我们称之为“tuple tree”，看起来如图1所示：

图1 示例tuple tree

在什么条件下，Storm才会认为一个从spout发送出来的消息被完整处理呢？答案就是下面的条件同时被满足：

• tuple tree不再生长
• 树中的任何消息被标识为“已处理”

如果在指定的时间内，一个消息衍生出来的tuple tree未被完全处理成功，则认为此消息未被完整处理。这个超时值可以通过任务级参数Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS 进行配置，默认超时值为30秒。

三、消息的生命周期

如果消息被完整处理或者未被完整处理，Storm会如何进行接下来的操作呢？为了弄清这个问题，我们来研究一下从spout发出来的消息的生命周期。这里列出了spout应该实现的接口：

首先， Storm使用spout实例的nextTuple()方法从spout请求一个消息（tuple）。 收到请求以后，spout使用open方法中提供的SpoutOutputCollector向它的输出流发送一个或多个消息。每发送一个消息，Spout会给这个消息提供一个message ID，它将会被用来标识这个消息。

假设我们从kestrel队列中读取消息，Spout会将kestrel 队列为这个消息设置的ID作为此消息的message ID。 向SpoutOutputCollector中发送消息格式如下：

接来下，这些消息会被发送到后续业务处理的bolts， 并且Storm会跟踪由此消息产生出来的新消息。当检测到一个消息衍生出来的tuple tree被完整处理后，Storm会调用Spout中的ack方法，并将此消息的messageID作为参数传入。同理，如果某消息处理超时，则此消息对应的Spout的fail方法会被调用，调用时此消息的messageID会被作为参数传入。

注意：一个消息只会由发送它的那个spout任务来调用ack或fail。如果系统中某个spout由多个任务运行，消息也只会由创建它的spout任务来应答（ack或fail），决不会由其他的spout任务来应答。

我们继续使用从kestrel队列中读取消息的例子来阐述高可靠性下spout需要做些什么(假设这个spout的名字是KestrelSpout)。

我们先简述一下kestrel消息队列：

当KestrelSpout从kestrel队列中读取一个消息，表示它“打开”了队列中某个消息。这意味着，此消息并未从队列中真正的删除，而是将此消息设置为“pending”状态，它等待来自客户端的应答，被应答以后，此消息才会被真正的从队列中删除。处于“pending”状态的消息不会被其他的客户端看到。另外，如果一个客户端意外的断开连接，则由此客户端“打开”的所有消息都会被重新加入到队列中。当消息被“打开”的时候，kestrel队列同时会为这个消息提供一个唯一的标识。

KestrelSpout就是使用这个唯一的标识作为这个tuple的messageID的。稍后当ack或fail被调用的时候，KestrelSpout会把ack或者fail连同messageID一起发送给kestrel队列，kestrel会将消息从队列中真正删除或者将它重新放回队列中。

四、可靠相关的API

为了使用Storm提供的可靠处理特性，我们需要做两件事情：

• 无论何时在tuple tree中创建了一个新的节点，我们需要明确的通知Storm；
• 当处理完一个单独的消息时，我们需要告诉Storm 这棵tuple tree的变化状态。