Storm累计求和中使用各种分组Grouping

       Shuffle Grouping: 随机分组， 随机派发stream里面的tuple， 保证bolt中的每个任务接收到的tuple数目相同.(它能实现较好的负载均衡)

       Fields Grouping：按字段分组， 比如按userid来分组， 具有同样userid的tuple会被分到同一任务， 而不同的userid则会被分配到不同的任务

       All Grouping： 广播发送,对于每一个tuple,Bolts中的所有任务都会收到.

       Global Grouping: 全局分组，这个tuple被分配到storm中的一个bolt的其中一个task.再具体一点就是分配给id值最低的那个task.

       Non Grouping: 随机分派，意思是说stream不关心到底谁会收到它的tuple.目前他和Shuffle grouping是一样的效果,

       Direct Grouping: 直接分组,这是一种比较特别的分组方法，用这种分组意味着消息的发送者具体由消息接收者的哪个task处理这个消息.只有被声明为Direct Stream的消息流可以声明这种分组方法.而且这种消息tuple必须使用emitDirect方法来发射.消息处理者可以通过TopologyContext来或者处理它的消息的taskid (OutputCollector.emit方法也会返回taskid)

 Fields Grouping 的代码

/**
 * 数字累加求和
 * 先添加storm依赖
 */
public class LocalTopologySumFieldsGrouping {
    /**
     * 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++,num%2));
            Utils.sleep(1000);
        }
        
        /**
         * 声明输出字段
         */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            //给values中的数据起个名字，方便后面的bolt从这个values中取数据
            //fields中定义的参数和values中传递的数值是一一对应的
            declarer.declare(new Fields("num","flag"));
        }
    }
    
    /**
     * 自定义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");
            System.out.println("线程id："+Thread.currentThread().getId()+",值："+value);
            //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());
        //.shuffleGrouping("spout1"); 表示让MyBolt接收MySpout发射出来的tuple
        topologyBuilder.setBolt("bolt1", new MyBolt(),3).fieldsGrouping("spout1", new Fields("flag"));
        
        //创建本地storm集群
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("sumTopology", new Config(), topologyBuilder.createTopology());
    }
    
}

ShuffleGrouping代码

/**
 * 数字累加求和
 * 先添加storm依赖
 */
public class LocalTopologySumShufferGrouping {
    /**
     * 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");
            System.out.println("线程id："+Thread.currentThread().getId()+",值："+value);//这样可以知道哪个线程接收到这个数据了.
            //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());
        //.shuffleGrouping("spout1"); 表示让MyBolt接收MySpout发射出来的tuple
        topologyBuilder.setBolt("bolt1", new MyBolt(),3).globalGrouping("spout1");
        
        //创建本地storm集群
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("sumTopology", new Config(), topologyBuilder.createTopology());
    }
}