标签：

看此博文前，建议先查看 

Storm批处理事务原理详解

为什么要进行批处理(Batch)？

逐个处理单个tuple，增加很多开销，如写库、输出结果频率过高

事务处理单个tuple效率比较低，因此storm中引入batch处理

批处理是一次性处理一批（batch）tuple,而事务则确保该批次要么全部处理成功，如果有处理失败的则全部不计，Storm会对失败的批次重新发送，且确保每个batch被且仅被处理一次

Spout有三种:

 分别为:

1. ITransactionalSpout<T>，同BaseTransactionalSpout<T>，普通事务Spout

2.IPartitionedTransactionalSpout<T>，同BasePartitionedTransactionalSpout<T>，分区事务Spout

3.IOpaquePartitionedTransactionalSpout<T>：同BaseOpaquePartitionedTransactionalSpout<T>，不透明分区事务Spout

Bolt有两种

1.IBatchBolt<T>：同BaseBatchBolt<T>，普通批处理

2.BaseTransactionalBolt：事务Bolt

实现接口Icommitter：标识IBatchBolt 或BaseTransactionalBolt是否是一个committer CoordinatedBolt

Spout

1.Spout： 普通事务Spout ITransactionalSpout

Interface ITransactionalSpout.Coordinator<X>:

Interface ITransactionalSpout.Emitter<X>

2.Spout: 分区事务IPartitionedTransactionalSpout<T>

分区事务Spout，主流事务Spout，原因是目前主流Message Queue都支持分区，分区的作用是增加MQ的吞吐量（每个分区作为一个数据源发送点），主流MQ如Kafka、RocketMQ

Interface IPartitionedTransactionalSpout.Coordinator

Interface IPartitionedTransactionalSpout.Emitter<X>

3.Spout : 不透明分区事务Spout

Interface IOpaquePartitionedTransactionalSpout

IOpaquePartitionedTransactionalSpout它不区分发新消息还是重发旧消息，全部用emitPartitionBatch搞定。虽然emitPartitionBatch返回的X应该是下一批次供自己使用的（emitPartitionBatch的第4个参数），但是只有一个批次成功以后X才会更新到ZooKeeper中，如果失败重发，emitPartitionBatch读取的X还是旧的。所以这时候自定义的X不需要记录当前批次的开始位置和下一批次的开始位置两个值，只需要记录下一批次开始位置一个值即可，例如：

public class BatchMeta { 
    public long  nextOffset; //下一批次的偏移量    
}

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。

我们以一个更简单的计算全局count的例子作说明，假设目前的统计结果为：

{ value = 4,

  prevValue = 1,

  txid = 2

}

BOLT

一.Interface IBatchBolt<T> 普通批处理

二 。Class BaseTransactionalBolt 事务批处理

 事务批处理与普通批处理的唯一区别:

 prepare的第四个参数

 IBatchBolt的prepare里面的最后一个参数是Object类型，而事务相关的是transactionAttempt

Storm批处理事务API详解

标签：

原文地址：http://www.cnblogs.com/thinkpad/p/5078610.html