【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========");
??
?}

?

?

?

?

?

?

?

?

?

【Storm入门级JAVA示例演示】

原文：http://gaojingsong.iteye.com/blog/2289928