jstorm系列-1：入门

public class TimeStampSpout implements IRichSpout{
    private static final long serialVersionUID = 1L;
    private SpoutOutputCollector collector;

    @Override
　　　　。。。。一切没有用到的函数不展示

    @Override
    public void nextTuple() {
        long now = System.currentTimeMillis();
        Values tuple = new Values(now);
        System.out.println("spout:"+tuple);
        this.collector.emit(tuple);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

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

}

public class GreetingBolt implements IRichBolt{
    private static final long serialVersionUID = 1L;
    private static Map<String, Integer> count = new TreeMap<String, Integer>(); 
    private final String morning = "morning"; 
    private final String noon = "noon"; 
    private final String afternoon = "afternoon"; 
    private final String evening = "evening";
    private static Integer total = 0;
    @Override
    public void cleanup() {
    }

    @Override
    public void execute(Tuple input) {
        long timestamp= input.getLong(0);
//        System.out.println("bolt:"+timestamp);
        total+=1;
//        获取十位数
        long second = (timestamp/10)%10;
        if(second<3){
            System.out.println("bolt:"+morning); 
            count.put(morning, (count.get(morning)==null)?1:count.get(morning)+1);
        }
        else if(second==3){
            System.out.println("bolt:"+noon);
            count.put(noon, (count.get(noon)==null)?1:count.get(noon)+1);
        }
        else if(second<8){
            System.out.println("bolt:"+afternoon);
            count.put(afternoon, (count.get(afternoon)==null)?1:count.get(afternoon)+1);
        }
        else{
            System.out.println("bolt:"+evening);
            count.put(evening, (count.get(evening)==null?1:count.get(evening)+1));
        }
        if(total%10==0){
            System.out.println("distribution show as followed:");
            System.out.println(morning+":"+1.0*((count.get(morning)==null)?0:count.get(morning))/total);
            System.out.println(noon+":"+1.0*((count.get(noon)==null?0:count.get(noon)))/total);
            System.out.println(afternoon+":"+1.0*((count.get(afternoon)==null?0:count.get(afternoon)))/total);
            System.out.println(evening+":"+1.0*((count.get(evening)==null?0:count.get(evening)))/total);
        }
    }

}

public class TestTopology {
    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("TimeStampSpout", new TimeStampSpout());
        builder.setBolt("GreetingBolt",new GreetingBolt()).shuffleGrouping("TimeStampSpout");
        LocalCluster cluster = new LocalCluster();
        Config config = new Config();
        cluster.submitTopology("test", config, builder.createTopology());
    }
}