一.storm入门demo的介绍
storm的入门helloworld有2种方式，一种是本地的，另一种是远程。
本地实现：
本地写好demo之后，不用搭建storm集群，下载storm的相关jar包即可实现storm的相关操作
远程实现：
本地写好demo之后，需要将其打成jar包，然后通过nimbus将jar包运行即可
本地打包注意事项：
由于打好的jar包会将其放到storm的集群上，因此storm上已经包含了运行的相关环境，但是在通过maven打包时需要storm-core设置成provided范围，不需要将storm-core的相关类打进jar包，以避免引起冲突
二.本地demo的实现
下载storm 所需jar包
1.spout数据源的实现
public class RandomStringSpout extends BaseRichSpout{ private final static Map map = new HashMap (); private SpoutOutputCollector collector; public RandomStringSpout(){ map.put(0, "kafka"); map.put(1, "nifi"); map.put(2, "flink"); map.put(3, "storm"); map.put(4, "spark"); } //在Spout组件初始化时被调用 public void open(Map arg0, TopologyContext topologyContextrg1, SpoutOutputCollector spoutOutputCollector) { System.err.println(" ============== open"); this.collector = spoutOutputCollector; } //nextTuple()方法是Spout实现的核心。 //也就是主要执行方法，用于输出信息,通过collector.emit方法发射 public void nextTuple() { //发送数据 collector.emit(new Values(map.get(ThreadLocalRandom.current().nextInt(4)))); try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); } } //用于声明数据格式,即输出的一个Tuple中，包含几个字段 public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("stream")); }}
2.Bolt数据过滤
public class WrapStarBolt extends BaseBasicBolt{ public void execute(Tuple tuple, BasicOutputCollector Collector) { String value = tuple.getStringByField("stream"); System.err.println("******"+value); } public void declareOutputFields(OutputFieldsDeclarer declarer) { //nothing to do }} public class WrapWellBolt extends BaseBasicBolt{ public void execute(Tuple tuple, BasicOutputCollector collector) { String value = tuple.getStringByField("stream"); System.err.println("#######"+value); } public void declareOutputFields(OutputFieldsDeclarer arg0) { //nothing to do }} 3.创建topology
//所有的spout bolt 会组成一个topology public class RadomStringTopologyLocal { public static void main(String[] args) throws InterruptedException { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("RandomStringSpout", new RandomStringSpout()); builder.setBolt("wrapStarBolt", new WrapStarBolt()).shuffleGrouping("RandomStringSpout"); builder.setBolt("wrapWellBolt", new WrapWellBolt()).shuffleGrouping("RandomStringSpout"); Config config = new Config(); config.setDebug(true); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("RadomStringTopologyLocal", config, builder.createTopology()); System.err.println("the first topology is start running at local"); TimeUnit.SECONDS.sleep(30); cluster.killTopology("RadomStringTopologyLocal"); cluster.shutdown(); }} //从运行的结果中可以看出写的demo已经运行 三.远程demo的实现
1.使用上一个demo中的RandomStringSpout WrapStarBolt WrapWellBolt 这三个类，然后编写新的RandomStringTopologyRemote类
public class RandomStringTopologyRemote { public static void main(String[] args) { final TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("RandomStringSpout", new RandomStringSpout()); builder.setBolt("WrapStarBolt", new WrapStarBolt(),4).shuffleGrouping("RandomStringSpout"); builder.setBolt("WrapWellBolt", new WrapWellBolt(),4).shuffleGrouping("RandomStringSpout"); final Config config = new Config(); config.setNumWorkers(3); try { StormSubmitter.submitTopology("RandomStringTopologyRemote", config, builder.createTopology()); } catch (Exception e) { e.printStackTrace(); } }}
2.把上面4个类通过maven工具打成jar包，并上传至nimbus所在的服务器上，上传完毕后通过以下命令远程启动storm
storm jar storm_test-0.0.1-SNAPSHOT.jar com.zpb.RandomStringTopologyRemote
远程启动storm的命令是：
storm jar + *.jar main函数的全路径名
3.通过UI工具查看
4.关闭提交的topology
storm kill 提交的topology名