storm源码分析---Transactional spouts

sadw

Trident是以小批量（batch）的形式在处理tuple，并且每一批都会分配一个唯一的transaction id。不同spout的特性不同，一个transactionalspout会有如下这些特性： 1、有着同样txid的batch一定是一样的。当重播一个txid对应的batch时，一定会重播和之前对应txid的batch中同样的tuples。
2、各个batch之间是没有交集的。每个tuple只能属于一个batch
3、每一个tuple都属于一个batch，无一例外
这是一类非常容易理解的spout， tuple 流被划分为固定的batch并且永不改变。（trident-kafka 有一个 transactional spout 的实现。）
你也许会问：为什么我们不总是使用transactional spout？这很容易理解。一个原因是并不是所有的地方都需要容错的。举例来说，TransactionalTridentKafkaSpout 工作的方式是一个batch包含的tuple来自某个kafka topic中的所有partition。一旦这个batch被发出，在任何时候如果这个batch被重新发出时，它必须包含原来所有的tuple以满足 transactional spout的语义。现在我们假定一个batch被TransactionalTridentKafkaSpout所发出，这个batch没有被成功处理，并且同时kafka的一个节点也down掉了。你就无法像之前一样重播一个完全一样的batch（因为kakfa的节点down掉，该topic的一部分partition可能会无法使用），整个处理会被中断。
这也就是"opaque transactional" spouts（不透明事务spout）存在的原因 - 他们对于丢失源节点这种情况是容错的，仍然能够帮你达到有且只有一次处理的语义。后面会对这种spout有所介绍。
在讨论"opaque transactional" spout之前，我们先来看看怎样为transactional spout设计一个具有exactly-once语义的State实现。这个State的类型是"transactionalstate" 并且它利用了任何一个txid总是对应同样的tuple序列这个语义。
假如说你有一个用来计算单词出现次数的topology，你想要将单词的出现次数以key/value对的形式存储到数据库中。key就是单词，value就是这个这个单词出现的次数。你已经看到只是存储一个数量是不足以知道你是否已经处理过一个batch的。你可以通过将value和txid一起存储到数据库中。这样的话，当更新这个count之前，你可以先去比较数据库中存储的txid和现在要存储的txid。如果一样，就跳过什么都不做，因为这个value之前已经被处理过了。如果不一样，就执行存储。这个逻辑可以工作的前提就是txid永不改变，并且Trident保证状态的更新是在batch之间严格顺序进行的。
考虑下面这个例子的运行逻辑，假定你在处理一个txid为3的包含下面tuple的batch：
["man"]
["man"]
["dog"]
假定数据库中当前保存了下面这样的key/value 对：
man => [count=3, txid=1]
dog => [count=4, txid=3]
apple => [count=10, txid=2]
单词“man”对应的txid是1. 因为当前的txid是3，你可以确定你还没有为这个batch中的tuple更新过这个单词的数量。所以你可以放心的给count加2并更新txid为3. 与此同时，单词“dog”的txid和当前的txid是相同的，因此你可以跳过这次更新。此时数据库中的数据如下：
man => [count=5, txid=3]
dog => [count=4, txid=3]
apple => [count=10, txid=2]