在Storm的Toplogy中設置多數據源Spout

上代碼:主要看main方法中的設置.   如下代碼是一般情況下的設置方法...Trident中設置多數據源看對應的博客總結

/**
 * 指定多個數據源
 * 數字累加求和
 * 先添加storm依賴
 */
public class LocalTopologyMeger {
    /**
     * spout需要繼承baserichspout，實現未實現的方法
     * @author Administrator
     *
     */
    public static class MySpout extends BaseRichSpout{
        private Map conf;
        private TopologyContext context;
        private SpoutOutputCollector collector;
        
        /**
         * 初始化方法，只會執行一次
         * 在這里面可以寫一個初始化的代碼
         * Map conf：其實里面保存的是topology的一些配置信息
         * TopologyContext context：topology的上下文，類似於servletcontext
         * SpoutOutputCollector collector：發射器，負責向外發射數據(tuple)
         */
        @Override
        public void open(Map conf, TopologyContext context,
                SpoutOutputCollector collector) {
            this.conf = conf;
            this.context = context;
            this.collector = collector;
        }

        int num = 1;
        /**
         * 這個方法是spout中最重要的方法，
         * 這個方法會被storm框架循環調用，可以理解為這個方法是在一個while循環之內
         * 每調用一次，會向外發射一條數據
         */
        @Override
        public void nextTuple() {
            System.out.println("spout發射："+num);
            //把數據封裝到values中，稱為一個tuple，發射出去
            this.collector.emit(new Values(num++));
            Utils.sleep(1000);
        }
        
        /**
         * 聲明輸出字段
         */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            //給values中的數據起個名字，方便后面的bolt從這個values中取數據
            //fields中定義的參數和values中傳遞的數值是一一對應的
            declarer.declare(new Fields("num"));
        }
        
    }
    
    
    /**
     * 自定義bolt需要實現baserichbolt
     * @author Administrator
     *
     */
    public static class MyBolt extends BaseRichBolt{
        private Map stormConf; 
        private TopologyContext context;
        private OutputCollector collector;
        
        /**
         * 和spout中的open方法意義一樣
         */
        @Override
        public void prepare(Map stormConf, TopologyContext context,
                OutputCollector collector) {
            this.stormConf = stormConf;
            this.context = context;
            this.collector = collector;
        }

        int sum = 0;
        /**
         * 是bolt中最重要的方法，當spout發射一個tuple出來，execute也會被調用，需要對spout發射出來的tuple進行處理
         */
        @Override
        public void execute(Tuple input) {
            //input.getInteger(0);//也可以根據角標獲取tuple中的數據
            Integer value = input.getIntegerByField("num");
            sum+=value;
            System.out.println("和："+sum);
        }
        
        /**
         * 聲明輸出字段
         */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            //在這沒必要定義了，因為execute方法中沒有向外發射tuple，所以就不需要聲明了。
            //如果nextTuple或者execute方法中向外發射了tuple，那么declareOutputFields必須要聲明，否則不需要聲明
        }
        
    }
    /**
     * 注意：在組裝topology的時候，組件的id在定義的時候，名稱不能以__開頭。__是系統保留的
     * @param args
     */
    public static void main(String[] args) {
        //組裝topology
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        topologyBuilder.setSpout("spout1", new MySpout());
        topologyBuilder.setSpout("spout2", new MySpout());
        //.shuffleGrouping("spout1"); 表示讓MyBolt接收MySpout發射出來的tuple
        topologyBuilder.setBolt("bolt1", new MyBolt()).shuffleGrouping("spout1").shuffleGrouping("spout2");
        
        //創建本地storm集群
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("sumTopology", new Config(), topologyBuilder.createTopology());
    }
}