【Storm入门级JAVA示例演示】

本例子主要演示，storm中的HelloWorld,从Spout不断的随机产生单词，然后在Bolt中进行组装输出。完整代码见附件。

详细代码如下：

// 实例化TopologyBuilder类
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        // 设置喷发节点并分配并发数，该并发数将会控制该对象在集群中的线程数。
        topologyBuilder.setSpout("SimpleSpout", new SimpleSpout(), 1);
        // 设置数据处理节点并分配并发数。指定该节点接收喷发节点的策略为随机方式。
        topologyBuilder.setBolt("SimpleBolt", new SimpleBolt(), 3).shuffleGrouping("SimpleSpout");
        Config config = new Config();
        config.setDebug(true);
        if (System.getProperty("os.name").contains("Windows")) {
          // 这里是本地模式下运行的启动代码。
            config.setMaxTaskParallelism(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("simple", config, topologyBuilder.createTopology());
        } else {
           config.setNumWorkers(1);
              StormSubmitter.submitTopology("simple", config, topologyBuilder.createTopology());
        }

 

// 实例化Spout类
String[] names = new String[]{"JAVA","PHP","IOS","ASP","HADOOP"};
 @Override
 public void open(Map conf, TopologyContext context,
   SpoutOutputCollector collector) {
  this.collector = collector;
  System.out.println("======Spout=====open========");
 }

 @Override
 public void nextTuple() {
  Utils.sleep(1000 * 3);
  Random rand = new Random();
  collector.emit(new Values(names[rand.nextInt(names.length)]));
  System.out.println("======Spout=====nextTuple========");
 }
 /**
     * 定义字段id，该id在简单模式下没有用处，但在按照字段分组的模式下有很大的用处。
     * 该declarer变量有很大作用，我们还可以调用declarer.declareStream();来定义stramId，该id可以用来定义更加复杂的流拓扑结构
     */
 @Override
 public void declareOutputFields(OutputFieldsDeclarer declarer) {
  declarer.declare(new Fields("msg"));
  System.out.println("======Spout=====declareOutputFields========");
  
 }

 

 

// 实例化Bolt类。
public void prepare(Map stormConf, TopologyContext context,
   OutputCollector collector) {
  System.out.println("======Bolt=====prepare========");
  
 }

 public void execute(Tuple input) {
  String desc = "out :"+ input.getString(0);
  System.out.println("====Bolt========desc:"+desc);
  
 }

 public void declareOutputFields(OutputFieldsDeclarer declarer) {
  System.out.println("====Bolt==declarer========");
  
 }