Apache Storm中内置了一种定时机制——tick,它能够让任何bolt的所有task每隔一段时间(精确到秒级,用户可以自定义)收到一个来自__systemd的__tick stream的tick tuple,bolt收到这样的tuple后可以根据业务需求完成相应的处理。
Tick功能从Apache Storm 0.8.0版本开始支持,本文在Apache Storm 0.9.1上测试。
在代码中如需使用tick,可以参照下面的方式:
若希望某个bolt每隔一段时间做一些操作,那么可以将bolt继承BaseBasicBolt/BaseRichBolt,并重写getComponentConfiguration()方法。在方法中设置Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS的值,单位是秒。
getComponentConfiguration()是backtype.storm.topology.IComponent接口中定义的方法, 在此方法的实现中可以定义以 ”Topology.*” 开头的此 bolt 特定的 Config 。
这样设置之后,此bolt的所有task都会每隔一段时间收到一个来自__systemd的__tick stream的tick tuple,因此execute()方法可以实现如下:
若希望Topology中的每个bolt都每隔一段时间做一些操作,那么可以定义一个Topology全局的tick,同样是设置Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS的值:
与Linux中的环境变量的优先级类似,storm中的tick也有优先级,即全局tick的作用域是全局bolt,但对每个bolt其优先级低于此bolt定义的tick。
这个参数的名字TOPOLOGY_TICK_TUPLE_FREQ_SECS具有一定的迷惑性,一眼看上去应该是Topology全局的,但实际上每个bolt也可以自己定义。
Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS是精确到秒级的。例如某bolt设置Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS为10s,理论上说bolt的每个task应该每个10s收到一个tick tuple。 实际测试发现,这个时间间隔的精确性是很高的,一般延迟(而不是提前)时间在 1ms 左右。
在bolt中的getComponentConfiguration()定义了该bolt的特定的配置后,storm框架会在TopologyBuilder.setBolt()方法中调用bolt的getComponentConfiguration()方法,从而设置该bolt的配置。
调用路径为:TopologyBuilder.setBolt()
-> TopologyBuilder.initCommon()
-> getComponentConfiguration()
测试使用的代码:
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>(); @Override public 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)); } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word", "count")); } @Override public 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{ @Override public void execute(Tuple tuple, BasicOutputCollector collector) { // 收到的tuple是tick tuple if (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时正常的tuple else{ collector.emit(new Values("a")); } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("test")); } @Override public 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(); } } }