direct(特供)
        只发送给指定的一个bolt.

        //a.通过emitDirect()方法发送元组
        //可以通过context.getTaskToComponent()方法得到所有taskId和组件名的映射
        collector.emitDirect(taskId,new Values(line));
        
        //b.指定directGrouping方式。
        builder.setBolt("split-bolt", new SplitBolt(),2).directGrouping("wcspout").setNumTasks(2);

package com.mao.storm.group.direct;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;

public class WordCountSpout implements IRichSpout {

    private TopologyContext context;
    private SpoutOutputCollector collector;
    List<String> stats;
    private int index = 0;
    public void open(Map map, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
        this.context = context;
        stats = new ArrayList<String>();
        stats.add("hollo world tom");
        stats.add("hollo world tom1");
        stats.add("hollo world tom2");
        List<Integer> tasks = context.getThisWorkerTasks();
    }
    private Random r = new Random();

    public void close() {

    }

    public void activate() {

    }

    public void deactivate() {

    }

    public void nextTuple() {
        int taskId = 0;
        Map<Integer,String> map = context.getTaskToComponent();
        for (Map.Entry<Integer,String> entry : map.entrySet()){
            if (entry.getValue().equals("split-bolt")){
                taskId = entry.getKey();
                break;
            }
        }
        if (index < 3) {
            String line = stats.get(r.nextInt(3));
            collector.emitDirect(taskId, new Values(line));
            index++;
        }
    }

    public void ack(Object o) {

    }

    public void fail(Object o) {

    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("line"));
    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

package com.mao.storm.group.direct;

import com.mao.storm.util.Util;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class SplitBolt implements IRichBolt {

    private TopologyContext context;
    private OutputCollector collector;

    public void prepare(Map map, TopologyContext context, OutputCollector collector) {
        this.context = context;
        this.collector = collector;
    }

    public void execute(Tuple tuple) {
        String line = tuple.getString(0);
        Util.sendToLocalhost(this,line);
        String[] arr = line.split(" ");
        for (String s : arr){
            collector.emit(new Values(s,1));
        }
    }

    public void cleanup() {

    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("word","count"));
    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

package com.mao.storm.group.direct;


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

/**
 * App
 */
public class App {

    public static void main(String[] args) throws Exception {

        TopologyBuilder builder = new TopologyBuilder();

        //设置spout
        builder.setSpout("wcSpout",new WordCountSpout()).setNumTasks(2);
        //设置creator-Bolt
        builder.setBolt("split-bolt",new SplitBolt(),2).directGrouping("wcSpout").setNumTasks(2);

        //设置countor-Bolt
     //   builder.setBolt("count-Bolt1",new WordCountBolt(),1).shuffleGrouping("split-bolt").setNumTasks(1);

        //设置countor-Bolt
      //  builder.setBolt("count-Bolt2",new WordCountBolt(),3).fieldsGrouping("count-Bolt1",new Fields("word")).setNumTasks(3);


        Config config = new Config();
        config.setNumWorkers(2);
        config.setDebug(true);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("wcShuffle",config,builder.createTopology());

        Thread.sleep(20000);

        cluster.shutdown();
        System.out.println("over");

    }
}