Storm高级原语（一）--Transactional topo

第一种设计缺点：一次只处理一个tuple，对数据库的大量调用，并且没有利用到storm的并行计算能力
第二种设计缺点：虽然使用了batch一次处理多个tuple，但仍旧没有利用到storm的并行计算能力
第三种设计很好地解决了这些问题
处理并行的方式是：transaction分为process和commit，可以有无数个process同时进行，但每次只能有一个commit在执行

MemoryTransactionalSpout spout = new MemoryTransactionalSpout(DATA, new Fields("word"), PARTITION_TAKE_PER_BATCH);
TransactionalTopologyBuilder builder = new TransactionalTopologyBuilder("global-count", "spout", spout, 3);
builder.setBolt("partial-count", new BatchCount(), 5)
        .shuffleGrouping("spout");
builder.setBolt("sum", new UpdateGlobalCount())
        .globalGrouping("partial-count");

TransactionalTopologyBuilder构造器中接受如下的参数：

一个transaction topology的id
spout在整个topology里面的id。
一个transactional spout。
一个可选的这个transactional spout的并行度。
topology的id是用来在zookeeper里面保存这个topology的当前进度状态的，所以如果你重启这个topology， 它可以接着前面的进度继续执行。

一个transaction topology里面有一个唯一的TransactionalSpout, 这个spout是通过TransactionalTopologyBuilder的构造函数来指定的。在这个例子里面，MemoryTransactionalSpout被用来从一个内存变量里面读取数据(DATA)。第二个参数指定spout发送的tuple的字段， 第三个参数指定每个batch的最大tuple数量。关于如何自定义TransactionalSpout我们会在后面介绍。

现在说说 bolts。这个topology并行地计算tuple的总数量。第一个bolt：BatchBolt，随机地把输入tuple分给各个task，然后各个task各自统计局部数量。第二个bolt：UpdateGlobalCount, 用全局grouping来汇总这个batch中tuple的数量，然后再更新到数据库里面的全局数量。

转自http://www.flyne.org/article/190