storm-workcount例子

pom.xml文件内容：

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

<groupId>com.xu.eshop</groupId>

<artifactId>storm-helloworld</artifactId>

<version>0.0.1-SNAPSHOT</version>

<name>storm-helloworld</name>

<description>storm-helloworld</description>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

<java.version>1.8</java.version>

</properties>

<groupId>junit</groupId>

<artifactId>junit</artifactId>

</dependency>

<groupId>org.apache.storm</groupId>

<artifactId>storm-core</artifactId>

</dependency>

<groupId>commons-collections</groupId>

<artifactId>commons-collections</artifactId>

</dependency>

</dependencies>

<build>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-shade-plugin</artifactId>

<exclude>META-INF/MSFTSIG.SF</exclude>

<exclude>META-INF/MSFTSIG.RSA</exclude>

</excludes>

</filter>

</filters>

</configuration>

<phase>package</phase>

<goals>

<goal>shade</goal>

</goals>

</transformer>

</transformers>

</configuration>

</execution>

</executions>

</plugin>

<groupId>org.codehaus.mojo</groupId>

<artifactId>exec-maven-plugin</artifactId>

<goals>

</goals>

</execution>

</executions>

<includePluginDependencies>false</includePluginDependencies>

<classpathScope>compile</classpathScope>

<mainClass>com.xu.eshop.storm.WordCountTopology</mainClass>

</configuration>

</plugin>

</plugins>

</build>

</project>

WordCountTopology.java代码：

package com.xu.eshop.storm;

import org.apache.storm.Config;

import org.apache.storm.LocalCluster;

import org.apache.storm.StormSubmitter;

import org.apache.storm.spout.SpoutOutputCollector;

import org.apache.storm.task.OutputCollector;

import org.apache.storm.task.TopologyContext;

import org.apache.storm.topology.OutputFieldsDeclarer;

import org.apache.storm.topology.TopologyBuilder;

import org.apache.storm.topology.base.BaseRichBolt;

import org.apache.storm.topology.base.BaseRichSpout;

import org.apache.storm.tuple.Fields;

import org.apache.storm.tuple.Tuple;

import org.apache.storm.tuple.Values;

import org.apache.storm.utils.Utils;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import java.util.HashMap;

import java.util.Map;

import java.util.Random;

public class WordCountTopology {

public static class RandomSentenceSpout extends BaseRichSpout {

private static final Logger LOGGER = LoggerFactory.getLogger(RandomSentenceSpout.class);

private Random random;

private SpoutOutputCollector collector;

public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector collector) {

this.collector = collector;

this.random = new Random();

}

public void nextTuple() {

Utils.sleep(100);

String[] sentences = new String[]{"the cow jumped over the moon", "an apple a day keeps the doctor away",

"four score and seven years ago", "snow white and the seven dwarfs", "i am at two with nature"};

final String sentence = sentences[random.nextInt(sentences.length)];

LOGGER.info("发射句子：" + sentence);

this.collector.emit(new Values(sentence));

}

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new Fields("sentence"));

}

public static class SplitSentence extends BaseRichBolt {

private OutputCollector collector;

public void prepare(Map conf, TopologyContext context, OutputCollector collector) {

this.collector = 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));

}

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new Fields("word"));

}

public static class WordCount extends BaseRichBolt {

private static final Logger LOGGER = LoggerFactory.getLogger(WordCount.class);

private OutputCollector collector;

private Map<String, Long> wordCounts = new HashMap<String, Long>();

public void prepare(Map conf, TopologyContext context, OutputCollector collector) {

this.collector = collector;

}

public void execute(Tuple tuple) {

String word = tuple.getStringByField("word");

Long count = this.wordCounts.get(word);

if(count == null) {

count = 0L;

}

count ++;

this.wordCounts.put(word, count);

LOGGER.info("单词计数" + word + "出现的次数是" + count);

this.collector.emit(new Values(word, count));

}

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new Fields("word", "count"));

}

public static void main(String[] args) {

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("RandomSentence", new RandomSentenceSpout(), 5);

builder.setBolt("SplitSentence", new SplitSentence(), 5)

.setNumTasks(10)

.shuffleGrouping("RandomSentence");

builder.setBolt("WordCount", new WordCount(), 10)

.setNumTasks(20)

.fieldsGrouping("SplitSentence", new Fields("word"));

Config config = new Config();

config.setDebug(false);

if(args != null && args.length > 0) {

config.setNumWorkers(3);

try {

StormSubmitter.submitTopology(args[0], config, builder.createTopology());

} catch (Exception e) {

e.printStackTrace();

}

} else {

config.setMaxTaskParallelism(20);

LocalCluster cluster = new LocalCluster();

cluster.submitTopology("WordCountTopology", config, builder.createTopology());

Utils.sleep(60000);

cluster.shutdown();

}