网站建设验收表,做销售网站那家好,用J2ee怎么做视频网站,动漫制作专业就业Spouts
Spout是Stream的消息产生源#xff0c;Spout组件的实现可以通过继承BaseRichSpout类或者其他Spout类来完成#xff0c;也可以通过实现IRichSpout接口来实现。 需要根据情况实现Spout类中重要的几个方法有#xff1a;
open方法
当一个Task被初始化的时候会调用此…
Spouts
Spout是Stream的消息产生源Spout组件的实现可以通过继承BaseRichSpout类或者其他Spout类来完成也可以通过实现IRichSpout接口来实现。 需要根据情况实现Spout类中重要的几个方法有
open方法
当一个Task被初始化的时候会调用此open方法。
一般都会在此方法中对发送Tuple的对象SpoutOutputCollector和配置对象TopologyContext初始化。 示例如下
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {this._collector collector;
}getComponentConfiguration方法
此方法用于声明针对当前组件的特殊的Configuration配置。 示例如下
public MapString, Object getComponentConfiguration() { if(!_isDistributed) {MapString, Object ret new HashMapString, Object(); ret.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 3); return ret;10 11 } else {return null;}
}这里便是设置了Topology中当前Component的线程数量上限。nextTuple方法
这是Spout类中最重要的一个方法。发射一个Tuple到Topology都是通过这个方法来实现的。 示例如下
public void nextTuple() { Utils.sleep(100); final String[] words new String[] {twitter,facebook,google}; final Random rand new Random(); final String word words[rand.nextInt(words.length)];collector.emit(new Values(word));
} 这里便是从一个数组中随机选取一个单词作为Tuple然后通过collector发送到Topology。declareOutputFields方法
此方法用于声明当前Spout的Tuple发送流。 Stream流的定义是通过OutputFieldsDeclare.declare方法完成的其中的参数包括了发送的域Fields。 示例如下
public void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields(word));
}另外除了上述几个方法之外还有ack、fail和close方法等;Storm在监测到一个Tuple被成功处理之后会调用ack方法处理失败会调用fail方法;这两个方法在BaseRichSpout等类中已经被隐式的实现了。—————————————————————————————————————————————— Bolts
Bolt类接收由Spout或者其他上游Bolt类发来的Tuple对其进行处理。
Bolt组件的实现可以通过继承BasicRichBolt类或者IRichBolt接口来完成。 Bolt类需要实现的主要方法有
prepare方法
此方法和Spout中的open方法类似为Bolt提供了OutputCollector用来从Bolt中发送Tuple。
Bolt中Tuple的发送可以在prepare方法中、execute方法中、cleanup等方法中进行一般都是些在execute中。
示例如下
public void prepare(Map conf, TopologyContext context, OutputCollector collector) {this._collector collector;
}getComponentConfiguration方法和Spout类一样在Bolt中也可以有getComponentConfiguration方法。 示例如下
public MapString, Object getComponentConfiguration() {MapString, Object conf new HashMapString, Object();conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, emitFrequencyInSeconds);return conf;
} 此例定义了从系统组件“_system”的“_tick”流中发送Tuple到当前Bolt的频率当系统需要每隔一段时间执行特定的处理时就可以利用这个系统的组件的特性来完成。execute方法
这是Bolt中最关键的一个方法对于Tuple的处理都可以放到此方法中进行。
具体的发送也是在execute中通过调用emit方法来完成的。 有两种情况一种是emit方法中有两个参数另一个种是有一个参数。
(1)emit有一个参数此唯一的参数是发送到下游Bolt的Tuple此时由上游发来的旧的Tuple在此隔断新的Tuple和旧的Tuple不再属于同一棵Tuple树。新的Tuple另起一个新的Tuple树。
(2)emit有两个参数第一个参数是旧的Tuple的输入流第二个参数是发往下游Bolt的新的Tuple流。此时新的Tuple和旧的Tuple是仍然属于同一棵Tuple树即如果下游的Bolt处理Tuple失败则会向上传递到当前Bolt当前Bolt根据旧的Tuple流继续往上游传递申请重发失败的Tuple。保证Tuple处理的可靠性。 declareOutputFields方法
用于声明当前Bolt发送的Tuple中包含的字段和Spout中类似。 示例如下
public void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields(obj,count,actualWindowLengthInSeconds));
} 此例说明当前Bolt类发送的Tuple包含了三个字段obj, count, actualWindowLengthInSeconds。—————————————————————————————————————————————— topology —————————————————————————————————————————————— Topology运行机制(1)Storm提交后会把代码首先存放到Nimbus节点的inbox目录下之后会把当前Storm运行的配置生成一个stormconf.ser文件放到Nimbus节点的stormdist目录中在此目录中同时还有序列化之后的Topology代码文件 (2)在设定Topology所关联的Spouts和Bolts时可以同时设置当前Spout和Bolt的executor数目和task数目默认情况下一个Topology的task的总和是和executor的总和一致的。之后系统根据worker的数目尽量平均的分配这些task的执行。worker在哪个supervisor节点上运行是由storm本身决定的 (3)任务分配好之后Nimbes节点会将任务的信息提交到zookeeper集群同时在zookeeper集群中会有workerbeats节点这里存储了当前Topology的所有worker进程的心跳信息 (4)Supervisor节点会不断的轮询zookeeper集群在zookeeper的assignments节点中保存了所有Topology的任务分配信息、代码存储目录、任务之间的关联关系等Supervisor通过轮询此节点的内容来领取自己的任务启动worker进程运行 (5)一个Topology运行之后就会不断的通过Spouts来发送Stream流通过Bolts来不断的处理接收到的Stream流Stream流是无界的。 最后一步会不间断的执行除非手动结束Topology。 Topology中的Stream处理时的方法调用过程如下 有几点需要说明的地方
(1)每个组件(Spout或者Bolt)的构造方法和declareOutputFields方法都只被调用一次。
(2)open方法、prepare方法的调用是多次的。入口函数中设定的setSpout或者setBolt里的并行度参数指的是executor的数目是负责运行组件中的task的线程的数目此数目是多少上述的两个方法就会被调用多少次在每个executor运行的时候调用一次。相当于一个线程的构造方法。
(3)nextTuple方法、execute方法是一直被运行的nextTuple方法不断的发射TupleBolt的execute不断的接收Tuple进行处理。只有这样不断地运行才会产生无界的Tuple流体现实时性。相当于线程的run方法。
(4)在提交了一个topology之后Storm就会创建spout/bolt实例并进行序列化。之后将序列化的component发送给所有的任务所在的机器(即Supervisor节 点)在每一个任务上反序列化component。
(5)Spout和Bolt之间、Bolt和Bolt之间的通信是通过zeroMQ的消息队列实现的。
(6)上图没有列出ack方法和fail方法在一个Tuple被成功处理之后需要调用ack方法来标记成功否则调用fail方法标记失败重新处理这个Tuple。 终止Topology
通过在Nimbus节点利用如下命令来终止一个Topology的运行
bin/storm kill topologyName kill之后可以通过UI界面查看topology状态会首先变成KILLED状态在清理完本地目录和zookeeper集群中的和当前Topology相关的信息之后此Topology就会彻底消失
