Storm 学习笔记（三）菜鸟乐园

朱雀 2022-05-25 07:45 153阅读 0赞

> #### [    经过前面两次学习笔记的介绍相信大家已经对storm的基本运行原理有了初步认识，][_storm] ####
> 
> #### [下面我们通过一个简单的小例子理解下：   ][_storm] ####

需求：做一个wordsum的程序

1+2+3+4+5.............

环境 ： eclipse mar

apache-storm-0.9.4

链接：  所有需要的jar都已经为大家准备好了放在lib下，大家下回去，使用时千万别忘了buildpath下，源码包一起提供，大家没事也可以跟进去看看  https://pan.baidu.com/s/1wWQq6ZzO5acHIAi4ivw-DQ

按照步骤走 Spout->Bolt->test测试

创建WsSpout类，代码如下

这里不用考虑变量i累加的问题，因为进程会自动一直发送

package com.zxy.storm;
    
    import java.util.List;
    import java.util.Map;
    
    
    import backtype.storm.spout.SpoutOutputCollector;
    import backtype.storm.task.TopologyContext;
    import backtype.storm.topology.OutputFieldsDeclarer;
    import backtype.storm.topology.base.BaseRichSpout;
    import backtype.storm.tuple.Fields;
    import backtype.storm.tuple.Values;
    
    public class WsSpout extends BaseRichSpout
    {
    
    	
    	private static final long serialVersionUID = 1L;
    	Map map;
    	TopologyContext context;
    	SpoutOutputCollector collector;
    	int i=0;
    
    	@Override 
    	public void nextTuple()
    	{	
    		i++;
    		List list = new Values(i);
    		collector.emit(list);
    		try
    		{
    			Thread.sleep(1000);
    		}
    		catch (InterruptedException e)
    		{
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.err.println("spout emit +++++++"+ list);
    		
    		
    	}
    
    	@Override
    	public void open(Map map, TopologyContext context, SpoutOutputCollector collector)
    	{
    		this.map=map;
    		this.context=context;
    		this.collector=collector;
    		
    	}
    
    	@Override
    	public void declareOutputFields(OutputFieldsDeclarer declarer)
    	{
    		//n是随便定义的一个字段名字，便于后面的Bolt接收
    		declarer.declare(new Fields("n"));
    	}
    	
    }

值得注意的是

List list = new Values(i);
    collector.emit(list);

collector.emit方法需要的是一个Tuple类型的参数，但是Tuple是一个接口不能实例化，这里实际封装数据的是

Values，而Values又继承了ArrayList，所以为了后期更灵活这里用了多态的写法

List list = new Values(i);

创建WsBolt类，代码如下

package com.zxy.storm;
    
    import java.util.Map;
    import backtype.storm.task.OutputCollector;
    import backtype.storm.task.TopologyContext;
    import backtype.storm.topology.OutputFieldsDeclarer;
    import backtype.storm.topology.base.BaseRichBolt;
    import backtype.storm.tuple.Tuple;
    
    public class WsBolt extends BaseRichBolt
    {
    
    	Map stormConf;
    	private TopologyContext context;
    	private OutputCollector collector;
    	int sum=0;
    
    	@Override
    	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
    	{
    		
    		this.stormConf=stormConf;
    		this.context=context;	
    		this.collector=collector;
    		
    	}
    
    	@Override
    	public void execute(Tuple input)
    	{
    		//接受数据
    		int number = input.getIntegerByField("n");
    		//处理数据
    		sum+=number;
    		//发送数据
    		System.err.println("sum======="+sum);
    	}
    
    	@Override
    	public void declareOutputFields(OutputFieldsDeclarer declarer)
    	{
    		//没有下一个Bolt了
    		
    	}
    
    }

创建Test类，代码如下

package com.zxy.storm;
    
    import backtype.storm.Config;
    import backtype.storm.LocalCluster;
    import backtype.storm.topology.TopologyBuilder;
    
    
    public class Test
    {
    	public static void main(String[] args)
    	{
    		TopologyBuilder tb=new TopologyBuilder();
    		tb.setSpout("ws", new WsSpout());
    		tb.setBolt("wsbolt",new WsBolt() ).shuffleGrouping("ws");
    		//本地模式,测试
    		LocalCluster localCluster = new LocalCluster();
    		localCluster.submitTopology("wordsum",new Config(), tb.createTopology());
    		
    	}
    
    }

执行结果如下        因为sleep了一下所以emit和sum结果显示顺序不固定

6697 \[Thread-11-wsb\] INFO  backtype.storm.daemon.executor - Prepared bolt wsb:(3)  
sum=======1  
6712 \[Thread-13-\_\_system\] INFO  backtype.storm.daemon.executor - Preparing bolt \_\_system:(-1)  
6716 \[Thread-13-\_\_system\] INFO  backtype.storm.daemon.executor - Prepared bolt \_\_system:(-1)  
6729 \[Thread-15-\_\_acker\] INFO  backtype.storm.daemon.executor - Preparing bolt \_\_acker:(1)  
6731 \[Thread-15-\_\_acker\] INFO  backtype.storm.daemon.executor - Prepared bolt \_\_acker:(1)  
spout emit +++++++\[1\]  
sum=======3  
spout emit +++++++\[2\]  
sum=======6  
spout emit +++++++\[3\]  
sum=======10  
spout emit +++++++\[4\]  
sum=======15  
spout emit +++++++\[5\]  
sum=======21  
spout emit +++++++\[6\]  
sum=======28  
spout emit +++++++\[7\]  
sum=======36

spout emit +++++++\[8\]

代码写的比较烂大家见谅，后面再为大家介绍下   flume+kafka+storm的应用这个非常实用，一般做个ETL或long还是很不错的

但是之前storm可能还需要介绍下其他相关知识，大家慢慢体会下

sq11W1WWQ

[_storm]: https://blog.csdn.net/paulfrank_zhang/article/details/80255554