storm源码分析

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class SimpleTopology {
    public static class SimpleSpout extends BaseRichSpout {
        private SpoutOutputCollector collector;

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

        @Override
        public void nextTuple() {
            collector.emit(new Values("Hello, Storm!"));
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

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

    public static class SimpleBolt extends BaseBasicBolt {
        @Override
        public void execute(Tuple input, BasicOutputCollector collector) {
            String message = input.getStringByField("message");
            System.out.println("Received: " + message);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            // No output fields
        }
    }

    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("simple-spout", new SimpleSpout(), 1);
        builder.setBolt("simple-bolt", new SimpleBolt(), 1).shuffleGrouping("simple-spout");

        Config config = new Config();
        config.setNumWorkers(1);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("simple-topology", config, builder.createTopology());
    }
}