storm_初识

<h3>简介:</h3>
Storm是一个分布式的，可靠的，容错的数据流处理系统。它会把工作任务委托给不同类型的组件，每个组件负责处理一项简单特定的任务。Storm集群的输入流由一个被称作spout的组件管理，spout把数据传递给bolt，bolt要么把数据保存到某种存储器，要么把数据传递给其它的bolt。你可以想象一下，一个Storm集群就是在一连串的bolt之间转换spout传过来的数据。
<h4>Storm的特性</h4>
在所有这些设计思想与决策中，有一些非常棒的特性成就了独一无二的Storm。
<li><b>简化编程</b></li> 如果你曾试着从零开始实现实时处理，你应该明白这是一件多么痛苦的事情。使用Storm，复杂性被大大降低了。使用一门基于JVM的语言开发会更容易，但是你可以借助一个小的中间件，在Storm上使用任何语言开发。有现成的中间件可供选择，当然也可以自己开发中间件。
<li><b>容错</b></li> Storm集群会关注工作节点状态，如果宕机了必要的时候会重新分配任务。
可扩展 所有你需要为扩展集群所做的工作就是增加机器。Storm会在新机器就绪时向它们分配任务。
<li><b>可靠</b></li> 所有消息都可保证至少处理一次。如果出错了，消息可能处理不只一次，不过你永远不会丢失消息。
<li><b>快速</b></li> 速度是驱动Storm设计的一个关键因素
<li><b>事务性</b></li> You can get exactly once messaging semantics for pretty much any computation.你可以为几乎任何计算得到恰好一次消息语义。
<h4>storm分布式计算结构</h4>
Storm 分布式计算结构称为topology(拓扑)，由stream(数据流)、spout(数据流的生成者)、bolt(运算)组成,如下图。
<pre>

Paste_Image.png

</pre>

<h3>storm本地demo搭建：</h3>
作为一个程序员来说，最大的毛病就是一言不合就上代码，好吧。我们就枚举一个官方用的比较多的例子吧，spot来喷发字符串，bolt1来以空格来分隔字符串继续向后续的计算模块bolt2分发,bolt2来通过来收集相同字符出现次数继续向计算模块bolt3分发，然后bolt3收集blot2的结果最终打印结果手动结束。如下图：

Paste_Image.png

新建pom项目，添加pom依赖:
<pre>
<dependencies>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>0.9.5</version>
</dependency>
</dependencies>
</pre>
SentenceSpout--单词生成类:
<pre>
public class SentenceSpout extends BaseRichSpout {
//用来发射数据的工具类
private SpoutOutputCollector collector;
private String[] sentences = {"my dog has fleas","i like cold beverages","the dog ate my homework","don't have a cow man","i don't think i like fleas"};
private int index = 0;
//每调用一次就可以向storm集群中发射一条数据（一个tuple元组），该方法会被不停的调用
public void nextTuple() {
this.collector.emit(new Values(sentences[index]));
index ++;
if(index >= sentences.length){
index = 0;
}
try {
Thread.sleep(1);
} catch (InterruptedException e) {
}
}
//初始化collector
@SuppressWarnings("rawtypes")
public void open(Map config, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
}
// 定义字段id，该id在简单模式下没有用处，但在按照字段分组的模式下有很大的用处。 该declarer变量有很大作用，我们还可以调用declarer.declareStream();来定义stramId，该id可以用来定义更加复杂的流拓扑结构
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("sentence"));
}
}
</pre>
SplitSentenceBolt--单词分隔类
<pre>
@SuppressWarnings("serial")
public class SplitSentenceBolt extends BaseRichBolt {
//用来定义继续向后续的计算模块发射数据的工具类
private OutputCollector collector;
public void execute(Tuple tuple) {
String sentence = tuple.getStringByField("sentence");
String[] words = sentence.split(" ");
for(String word : words){
this.collector.emit(new Values(word));
}
}
//初始化
@SuppressWarnings("rawtypes")
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}
</pre>
WordCountBolt--相同的单词统计
<pre>
public class WordCountBolt extends BaseRichBolt {
private OutputCollector collector;
private HashMap<String, Long> counts = null;
public void execute(Tuple tuple) {
String word = tuple.getStringByField("word");
Long count = this.counts.get(word);
if (count == null) {
count = 1L;
}
count++;
this.counts.put(word, count);
this.collector.emit(new Values(word, count));
}
@SuppressWarnings("rawtypes")
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
this.collector = collector;
counts = new HashMap<>();
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word","count"));
}
}
</pre>
BaseRichBolt--统计单词结果
<pre>
@SuppressWarnings("serial")
public class ReportBolt extends BaseRichBolt {
private HashMap<String, Long> counts = null;
@Override
public void execute(Tuple tuple) {
String word = tuple.getStringByField("word");
Long count = tuple.getLongByField("count");
this.counts.put(word, count);
}
@SuppressWarnings("rawtypes")
@Override
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
counts = new HashMap<>();
}
@Override
public void declareOutputFields(OutputFieldsDeclarer arg0) {
}
public void cleanup(){
System.err.println("---final counts---");
List<String> keys = new ArrayList<>();
keys.addAll(counts.keySet());
Collections.sort(keys);
for(String key : keys){
System.err.println(key + " : " + this.counts.get(key));
}
System.err.println("---end---");
}
}
</pre>
最后的最后我要上main方法啦：
<pre>
public class WordCountTopology {
private static final String SPOUT = "spout";
private static final String SPLIT_BOLT = "splitBolt";
private static final String COUNT_BOLT = "countBolt";
private static final String REPORT_BOLT = "reportBolt";
private static final String TOPOLOGY_NAME = "wordCountTopology";
public static void main(String[] args) throws InterruptedException {
//数据发射器
SentenceSpout spout = new SentenceSpout();
//字符串分隔计算模块
SplitSentenceBolt splitBolt = new SplitSentenceBolt();
//字符串统计模块
WordCountBolt countBolt = new WordCountBolt();
//最后打印模块
ReportBolt reportBolt = new ReportBolt();
//创建Topology
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SPOUT, spout);
builder.setBolt(SPLIT_BOLT, splitBolt).shuffleGrouping(SPOUT);
builder.setBolt(COUNT_BOLT, countBolt).fieldsGrouping(SPLIT_BOLT, new Fields("word"));
builder.setBolt(REPORT_BOLT, reportBolt).globalGrouping(COUNT_BOLT);
Config config = new Config();
//定义本地storm集群，如果放在linux虚拟机上跑略有不同
LocalCluster cluster = new LocalCluster();
cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());
Thread.sleep(10000);
//kill Topology,当Topology启动以后会一直执行直到kill Topology
cluster.killTopology(TOPOLOGY_NAME);
//关闭集群，这个方法跟redis的集群关闭一样
cluster.shutdown();
}
}
</pre>
最后统计结果：
<pre>
---final counts---
a : 1553
ate : 1554
beverages : 1554
cold : 1554
cow : 1553
dog : 3107
don't : 3105
fleas : 3106
has : 1554
have : 1553
homework : 1554
i : 4658
like : 3106
man : 1553
my : 3107
the : 1554
think : 1553
---end---
</pre>
如果说这么就结束了，是不是太快了啊，来分析分析BaseRichSpout、BaseRichBolt代码结构吧：
<pre>

Paste_Image.png

</pre>
<p>关于BaseRichSpout的ack和fail这两个方法我不得不讲一下，因为是可以用在Tuple Stream传递后确认成功和失败，当成功以后可以打印成功的log失败以后可以在fail方法中定义重发.
</p>

<pre>

Paste_Image.png

</pre>
<p>关于BaseRichBolt的cleanup定义一个bolt结束时被执行，但是不能保证被执行。</p>
<p>BaseComponent类的存在也就是为了隔出一层出来实现getComponentConfiguration避免让更多的不需要的子类累赘实现这个方法。</p>
<p>总的来说storm暴露给我们常用的这几个类以及其结构还是很简单的，跟我们自身自己写代码的层次差不多，很容易理解。</p>
storm成长之路_初识就讲到这里啦，谢谢！