轉(zhuǎn)載自kqdongnanf-博客園；Email：kqdongnanf@yahoo.com。

1. tick的功能

Apache Storm中內(nèi)置了一種定時機制——tick，它能夠讓任何bolt的所有task每隔一段時間（精確到秒級，用戶可以自定義）收到一個來自__systemd的__tick stream的tick tuple，bolt收到這樣的tuple后可以根據(jù)業(yè)務(wù)需求完成相應(yīng)的處理。

Tick功能從Apache Storm 0.8.0版本開始支持，本文在Apache Storm 0.9.1上測試。

2. 在代碼中使用tick及其作用

在代碼中如需使用tick，可以參照下面的方式：

2.1. 為bolt設(shè)置tick

若希望某個bolt每隔一段時間做一些操作，那么可以將bolt繼承BaseBasicBolt/BaseRichBolt，并重寫getComponentConfiguration()方法。在方法中設(shè)置Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS的值，單位是秒。

getComponentConfiguration()是backtype.storm.topology.IComponent接口中定義的方法，在此方法的實現(xiàn)中可以定義以”Topology.*”開頭的此bolt特定的Config。

這樣設(shè)置之后，此bolt的所有task都會每隔一段時間收到一個來自__systemd的__tick stream的tick tuple，因此execute()方法可以實現(xiàn)如下：

2.2. 為Topology全局設(shè)置tick

若希望Topology中的每個bolt都每隔一段時間做一些操作，那么可以定義一個Topology全局的tick，同樣是設(shè)置Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS的值：

2.3. tick設(shè)置的優(yōu)先級

與Linux中的環(huán)境變量的優(yōu)先級類似，storm中的tick也有優(yōu)先級，即全局tick的作用域是全局bolt，但對每個bolt其優(yōu)先級低于此bolt定義的tick。

這個參數(shù)的名字TOPOLOGY_TICK_TUPLE_FREQ_SECS具有一定的迷惑性，一眼看上去應(yīng)該是Topology全局的，但實際上每個bolt也可以自己定義。

2.4. tick的精確度

Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS是精確到秒級的。例如某bolt設(shè)置Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS為10s，理論上說bolt的每個task應(yīng)該每個10s收到一個tick tuple。
實際測試發(fā)現(xiàn)，這個時間間隔的精確性是很高的，一般延遲（而不是提前）時間在1ms左右。測試環(huán)境：3臺虛擬機做supervisor，每臺配置：4Cpu、16G內(nèi)存、千兆網(wǎng)卡。

3. storm tick的實現(xiàn)原理

在bolt中的getComponentConfiguration()定義了該bolt的特定的配置后，storm框架會在TopologyBuilder.setBolt()方法中調(diào)用bolt的getComponentConfiguration()方法，從而設(shè)置該bolt的配置。

調(diào)用路徑為：TopologyBuilder.setBolt()-> TopologyBuilder.initCommon()-> getComponentConfiguration()

4. 附件

測試使用的代碼：

package storm.starter;import backtype.storm.Config; import backtype.storm.Constants; import backtype.storm.LocalCluster; import backtype.storm.StormSubmitter; import backtype.storm.task.ShellBolt; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.IRichBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.TopologyBuilder; import backtype.storm.topology.base.BaseBasicBolt; import backtype.storm.topology.base.BaseRichBolt; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; import storm.starter.spout.RandomSentenceSpout;import java.text.SimpleDateFormat; import java.util.Date; import java.util.HashMap; import java.util.Map;public class MyTickTestTopology {public static class WordCount extends BaseBasicBolt {Map<String, Integer> counts = new HashMap<String, Integer>();@Overridepublic void execute(Tuple tuple, BasicOutputCollector collector) {if (tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID)&& tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID)){System.out.println("################################WorldCount bolt: "+ new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(new Date()));}else{collector.emit(new Values("a", 1));}}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word", "count"));}@Overridepublic Map<String, Object> getComponentConfiguration() {Config conf = new Config();conf.put(conf.TOPOLOGY_TICK_TUPLE_FREQ_SECS,10);return conf;}}public static class TickTest extends BaseBasicBolt{@Overridepublic void execute(Tuple tuple, BasicOutputCollector collector) {// 收到的tuple是tick tupleif (tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID)&& tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID)){System.out.println("################################TickTest bolt: "+ new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(new Date()));}// 收到的tuple時正常的tupleelse{collector.emit(new Values("a"));}}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("test"));}@Overridepublic Map<String, Object> getComponentConfiguration() {Config conf = new Config();conf.put(conf.TOPOLOGY_TICK_TUPLE_FREQ_SECS,20);return conf;}}public static void main(String[] args) throws Exception {TopologyBuilder builder = new TopologyBuilder();builder.setSpout("spout", new RandomSentenceSpout(), 3);builder.setBolt("count", new WordCount(), 3).shuffleGrouping("spout");builder.setBolt("tickTest", new TickTest(), 3).shuffleGrouping("count");Config conf = new Config();conf.put(conf.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 7);conf.setDebug(false);if (args != null && args.length > 0) {conf.setNumWorkers(3);StormSubmitter.submitTopology(args[0], conf, builder.createTopology());}else {conf.setMaxTaskParallelism(3);LocalCluster cluster = new LocalCluster();cluster.submitTopology("word-count", conf, builder.createTopology());// Thread.sleep(10000); // cluster.shutdown();}} }

總結(jié)

以上是生活随笔為你收集整理的关于Storm Tick的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。