storm的定时任务

 

 

应用场景：

 

第一种方法

 

 参考代码StormTopologyTimer1.java

package yehua.storm;

import java.util.Map;

import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;


public class StormTopologyTimer1 {

public static class MySpout extends BaseRichSpout{
private Map conf;
private TopologyContext context;
private SpoutOutputCollector collector;
        @Override
public void open(Map conf, TopologyContext context,
                SpoutOutputCollector collector) {
this.conf = conf;
this.collector = collector;
this.context = context;
        }

int num = 0; 
        @Override
public void nextTuple() {
            num++;
            System.out.println("spout:"+num);
this.collector.emit(new Values(num));
            Utils.sleep(1000);
        }

        @Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("num"));
        }

    }



public static class MyBolt extends BaseRichBolt{

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

int sum = 0;
        @Override
public void execute(Tuple input) {
if(input.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID)){
                System.out.println("定时时间到了");
            }else{
                Integer num = input.getIntegerByField("num");
                sum += num;
                System.out.println("sum="+sum);
            }
        }

        @Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {

        }

    }



public static void main(String[] args) {
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        String spout_id = MySpout.class.getSimpleName();
        String bolt_id = MyBolt.class.getSimpleName();

        topologyBuilder.setSpout(spout_id, new MySpout());
        topologyBuilder.setBolt(bolt_id, new MyBolt()).shuffleGrouping(spout_id);


        Config config = new Config();
//设置定时任务
        config.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 10);//表示storm每隔10秒都会给topology里面的所有bolt发送一个系统级别的tuple
        String topology_name = StormTopologyTimer1.class.getSimpleName();
if(args.length==0){
//在本地运行
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology(topology_name, config, topologyBuilder.createTopology());
        }else{
//在集群运行
try {
                StormSubmitter.submitTopology(topology_name, config, topologyBuilder.createTopology());
            } catch (AlreadyAliveException e) {
                e.printStackTrace();
            } catch (InvalidTopologyException e) {
                e.printStackTrace();
            } catch (AuthorizationException e) {
                e.printStackTrace();
            }
        }

    }

}

 

 方法二：

 

所以我们应该这么干

这个方法也ok的，这个不同于前面的在main里设置定时，这个是在bolt里设置，这个方法就是针对bolt设置，针对某一个bolt设置定时任务！

 

参考代码StormTopologyTimer2.java

package yehua.storm;

import java.util.HashMap;
import java.util.Map;

import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;


public class StormTopologyTimer2 {

public static class MySpout extends BaseRichSpout{
private Map conf;
private TopologyContext context;
private SpoutOutputCollector collector;
        @Override
public void open(Map conf, TopologyContext context,
                SpoutOutputCollector collector) {
this.conf = conf;
this.collector = collector;
this.context = context;
        }

int num = 0; 
        @Override
public void nextTuple() {
            num++;
            System.out.println("spout:"+num);
this.collector.emit(new Values(num));
            Utils.sleep(1000);
        }

        @Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("num"));
        }

    }



public static class MyBolt extends BaseRichBolt{

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

int sum = 0;
        @Override
public void execute(Tuple input) {
if(input.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID)){
                System.out.println("定时时间到了");
            }else{
                Integer num = input.getIntegerByField("num");
                sum += num;
                System.out.println("sum="+sum);
            }
        }

        @Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {

        }

        @Override
public Map<String, Object> getComponentConfiguration() {
            HashMap<String, Object> hashMap = new HashMap<String, Object>();
            hashMap.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 10);//给当前bolt设置定时任务
return hashMap;
        }


    }



public static void main(String[] args) {
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        String spout_id = MySpout.class.getSimpleName();
        String bolt_id = MyBolt.class.getSimpleName();

        topologyBuilder.setSpout(spout_id, new MySpout());
        topologyBuilder.setBolt(bolt_id, new MyBolt()).shuffleGrouping(spout_id);


        Config config = new Config();
        String topology_name = StormTopologyTimer2.class.getSimpleName();
if(args.length==0){
//在本地运行
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology(topology_name, config, topologyBuilder.createTopology());
        }else{
//在集群运行
try {
                StormSubmitter.submitTopology(topology_name, config, topologyBuilder.createTopology());
            } catch (AlreadyAliveException e) {
                e.printStackTrace();
            } catch (InvalidTopologyException e) {
                e.printStackTrace();
            } catch (AuthorizationException e) {
                e.printStackTrace();
            }
        }

    }

}