STORM的三种事务

Spout和Bolt

Spout

[ITransactionalSpout][<T>]，同****BaseTransactionalSpout****<T>****，普通事务****Spout [IPartitionedTransactionalSpout<T>]，同****BasePartitionedTransactionalSpout****<T>****，分区事务****Spout [IOpaquePartitionedTransactionalSpout][<T>]：同****BaseOpaquePartitionedTransactionalSpout****<T>****，不透明分区事务****Spout

Bolt [IBatchBolt][<T>]：同****BaseBatchBolt****<T>****，普通批处理 [BaseTransactionalBolt]：事务Bolt

接口****Icommitter****：标识IBatchBolt 或[BaseTransactionalBolt]是否是一个****committer****[CoordinatedBolt] 1、传统事务

[ITransactionalSpout][<T>]：普通事务Spout — [ITransactionalSpout.Coordinator][<X>] —initializeTransaction(BigInteger txid, XprevMetadata) ： 创建一个新的metadata，当isReady 为true时，发射该metadata（事务tuple）到“batchemit”流 —isReady ：为true时启动新事务，需要时可以在此sleep — [ITransactionalSpout.Emitter][<X>] —[][emitBatch]([TransactionAttempt] tx,[X] coordinatorMeta, [BatchOutputCollector] collector) ：逐个发射batch的tuple

2、分区事务

[IPartitionedTransactionalSpout<T>]：分区事务****Spout****，主流事务****Spout****，原因是目前主流****MessageQueue****都支持分区，分区的作用是增加****MQ****的吞吐量（每个分区作为一个数据源发送点），主流****MQ****如****Kafka****、****RocketMQ ** —****[][IPartitionedTransactionalSpout.Coordinator]** ** —****[][isReady]()：同上 —[][numPartitions]**()：返回分区个数。当增加了数据源新分区，同时一个事务被replayed，此时则不发射新分区的tuples，因为它知道该事务中有多少个分区。

—[][IPartitionedTransactionalSpout.Emitter]<****[X]> —[emitPartitionBatchNew]([TransactionAttempt] tx,[BatchOutputCollector] collector,int partition,[X] lastPartitionMeta)：发射一个新的Batch，返回Metadata —[emitPartitionBatch]([TransactionAttempt] tx,[BatchOutputCollector] collector,int partition,[X] partitionMeta)：如果这批消息Bolt消费失败了，emitPartitionBatch负责重发这批消息 3、不透明分区事务

IOpaquePartitionedTransactionalSpout****<T>****：不透明分区事务****Spout ** —****[IOpaquePartitionedTransactionalSpout.Coordinator]** —[isReady]()：同上 —[IOpaquePartitionedTransactionalSpout.Emitter]<****[X]> — [emitPartitionBatch]([TransactionAttempt] tx,[BatchOutputCollector] collector,int partition,[X] lastPartitionMeta) — [numPartitions]()

它不区分发新消息还是重发旧消息，全部用emitPartitionBatch搞定。虽然emitPartitionBatch返回的X应该是下一批次供自己使用的（emitPartitionBatch的第4个参数），但是只有一个批次成功以后X才会更新到ZooKeeper中，如果失败重发，emitPartitionBatch读取的X还是旧的。所以这时候自定义的X不需要记录当前批次的开始位置和下一批次的开始位置两个值，只需要记录下一批次开始位置一个值即可，例如： public class BatchMeta{ public long nextOffset; //下一批次的偏移量 } 4、分布事务和不透明分区事务Spout的区别与联系

IPartitionedTransactionalSpout和IOpaquePartitionedTransactionalSpout都是把tuple封装成batch进行处理，同时可以保证每一个tuple都被完整地处理，都支持消息重发。为了支持事务性，它们为每一个批次（batch）提供一个唯一的事务ID（transaction id：txid），txid是顺序递增的，而且保证对批次的处理是强有序的，即必须完整处理完txid=1才能再接着处理txid=2。

二者的区别以及用法： IPartitionedTransactionalSpout的每一个tuple都会绑定在固定的批次中。无论一个tuple重发多少次，它都在同一个批次里面，都有同样的事务ID；一个tuple不会出现在两个以上的批次里。一个批次无论重发多少次，它也只有一个唯一且相同的事务ID，不会改变。这也就是说，一个批次无论重发多少次，它所包含的内容都是完全一致的。

但是IPartitionedTransactionalSpout会有一个问题，虽然这种问题非常罕见：假设一批消息在被bolt消费过程中失败了，需要spout重发，此时如果正巧遇到消息发送中间件故障，例如某一个分区不可读，spout为了保证重发时每一批次包含的tuple一致，它只能等待消息中间件恢复，也就是卡在那里无法再继续发送给bolt消息了，直至消息中间件恢复。IOpaquePartitionedTransactionalSpout可以解决这个问题。 而IOpaquePartitionedTransactionalSpout为了解决这个问题，它不保证每次重发一个批次的消息所包含的tuple完全一致。也就是说某个tuple可能第一次在txid=2的批次中出现，后面有可能在txid=5的批次中出现。这种情况只出现在当某一批次消息消费失败需要重发且恰巧消息中间件故障时。这时，IOpaquePartitionedTransactionalSpout不是等待消息中间件故障恢复，而是先读取可读的partition。例如txid=2的批次在消费过程中失败了，需要重发，恰巧消息中间件的16个分区有1个分区(partition=3)因为故障不可读了。这时候IOpaquePartitionedTransactionalSpout会先读另外的15个分区，完成txid=2这个批次的发送，这时候同样的批次其实包含的tuple已经少了。假设在txid=5时消息中间件的故障恢复了，那之前在txid=2且在分区partition=3的tuple会重新发送，包含在txid=5的批次中。 在使用IOpaquePartitionedTransactionalSpout时，因为tuple与txid的对应关系有可能改变，因此与业务计算结果同时保存一个txid就无法保证事务性了。这时候解决方案会稍微复杂一些，除了保存业务计算结果以外，还要保存两个元素：前一批次的业务计算结果以及本批次的事务ID。 如： {value = 4, prevValue = 1, txid = 2 }