本文共 1929 字,大约阅读时间需要 6 分钟。
本节书摘来自华章计算机《Storm企业级应用:实战、运维和调优》一书中的第3章,第3.1节,作者:马延辉 陈书美 雷葆华著, 更多章节内容可以访问云栖社区“华章计算机”公众号查看。
Tuple是Storm的主要数据结构,并且是Storm中使用的最基本单元、数据模型和元组。
3.1.1 Tuple描述Tuple就是一个值列表,Tuple中的值可以是任何类型的,动态类型的Tuple的f?ields可以不用声明;默认情况下,Storm中的Tuple支持私有类型、字符串、字节数组等作为它的字段值,如果使用其他类型,就需要序列化该类型。Tuple的字段默认类型有:integer、float、double、long、short、string、byte、binary(byte[])。Tuple数据结构如图3-1所示。 Tuple可以理解成键值对。例如,创建一个Bolt要发送两个字段(命名为double和triple),其中键就是定义在declareOutputFields方法中的Fields对象,值就是在emit方法中发送的Values对象。以下是一个简单例子。public class DoubleAndTripleBolt extends BaseRichBolt {OutputCollectorBase _collector;@Overridepublic void prepare(Map conf, TopologyContext context, OutputCollectorBase collector) { _collector = collector;}@Overridepublic void execute(Tuple input) { int val = input.getInteger(0); _collector.emit(input, new Values(val*2, val*3)); _collector.ack(input);}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("double", "triple"));}}
此外,在使用的Storm Java包中,backtype.storm.tuple主要有以下几个类:
Fileds.class。MessageId.class。Tuple.class。TupleImpl.class。Values.class。列出以上内容是为了更好地理解Tuple,这样能够从本质上理解Tuple,在使用时更加得心应手。3.1.2 Tuple的生命周期了解一个Tuple的生命周期就需要查看源码,如下的Java代码展示了Spout(消息源)接口发出Tuple(消息)的整个过程。public interface ISpout extends Serializable { void open(Map conf, TopologyContext context, SpoutOutputCollector collector); void nextTuple(); void ack(Object msgId); void fail(Object msgId); void close(); }
首先,Storm调用Spout(消息源)的nextTuple方法来获取下一个Tuple,Spout通过Open方法的参数提供的SpoutOutputCollector将新Tuple发射到其中一个输出消息流。
发射Tuple时,Spout提供一个message-id,通过这个ID来追踪该Tuple。接下来,Storm跟踪该Tuple的树形结构是否成功创建,并根据messageid调用Spout中的ack函数,以确认Tuple是否被完全处理。如果Tuple超时,则调用Spout的fail方法。由此看出,同一个Tuple不管是acked,还是failed都是由创建它的Spout发出并维护的,所以,即使Spout在集群环境中同时执行很多的任务,该Tuple也不会被其他任务调用或生成acked或failed状态。总之,Storm会利用内部的Acker机制保证每个Tuple被可靠地处理。最后,在任务完成后,Spout调用Close方法结束Tuple的使命。转载地址:http://xotkx.baihongyu.com/