Storm Bolt之定时机制Tick应用

       Storm中有一种内置的定时机制Storm Bolt之Tick，可以在任何bolt的task每个一定时间（支持通过用户自定义配置）收到来自System Id的tick tuple。Bolt在收到这样的tuple后，根据业务需求完成相应的处理流程。

     在构建topology时，调用addConfiguration()方法，完成对tick时间间隔的设置，如下：

public static void main(String[] args) throws Exception {
        //开始组合bolt和spout的组件
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("Test_Spout", new TestSpout(TopoLoad.getKafkaSpoutConfig()), conf.getInt(SPOUT_PARALLELISN_HINT_INT));
        builder.setBolt("Test_Bolt", new TestBolt(), 4).localOrShuffleGrouping("Test_Spout")
        .addConfiguration(TOPOLOGY_TICK_TUPLE_FREQ_SECS,10);
}
           

 在具体的Bolt中，用户的Bolt需要继承BaseRichBolt/BaseBasicBolt，在Bolt中重写getComponentConfiguration()方法，设置tick时间间隔。并在bolt的execute(Tuple tuple)方法中调用TupleUtils.isTick(input)判断tuple是否为tick tuple，相应进行业务处理。

public class TestBolt extends BaseRichBolt {
    int flushIntervalSecs = 1;
    int batchSize = 1000;
    private String streamFieldName;
    OutputCollector collector;

    public TestBolt(String streamFieldName) {
        this.streamFieldName = streamFieldName;
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        if (!TupleUtils.isTick(input)) {//如果收到的tuple不是tick的tuple，不处理
             //业务处理
        }else{
             //业务处理
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return TupleUtils.putTickFrequencyIntoComponentConfig(super.getComponentConfiguration(), flushIntervalSecs);
    }

    public MySqlBolt withBatchSize(int batchSize) {
        this.batchSize = batchSize;
        return this;
    }

    public MySqlBolt withFlushIntervalSecs(int flushIntervalSecs) {
        this.flushIntervalSecs = flushIntervalSecs;
        return this;
    }
}