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">

<modelVersion>4.0.0</modelVersion>

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

<artifactId>storm-helloworld</artifactId>

<version>0.0.1-SNAPSHOT</version>

<packaging>jar</packaging>

<name>storm-helloworld</name>

<description>storm-helloworld</description>

<properties>

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

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

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

</properties>

<dependencies>

<dependency>

<groupId>junit</groupId>

<artifactId>junit</artifactId>

<version>4.10</version>

<scope>test</scope>

</dependency>

<dependency>

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

<artifactId>storm-core</artifactId>

<version>1.1.0</version>

<!--<scope>provided</scope>-->

</dependency>

<dependency>

<groupId>commons-collections</groupId>

<artifactId>commons-collections</artifactId>

<version>3.2.1</version>

</dependency>

</dependencies>

<build>

<sourceDirectory>src/main/java</sourceDirectory>

<testSourceDirectory>src/test/java</testSourceDirectory>

<plugins>

<plugin>

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

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

<configuration>

<createDependencyReducedPom>true</createDependencyReducedPom>

<filters>

<filter>

<artifact>*:*</artifact>

<excludes>

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

<exclude>META-INF/*.sf</exclude>

<exclude>META-INF/*.DSA</exclude>

<exclude>META-INF/*.dsa</exclude>

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

<exclude>META-INF/*.rsa</exclude>

<exclude>META-INF/*.EC</exclude>

<exclude>META-INF/*.ec</exclude>

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

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

</excludes>

</filter>

</filters>

</configuration>

<executions>

<execution>

<phase>package</phase>

<goals>

<goal>shade</goal>

</goals>

<configuration>

<transformers>

<transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />

<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">

</transformer>

</transformers>

</configuration>

</execution>

</executions>

</plugin>

<plugin>

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

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

<version>1.2.1</version>

<executions>

<execution>

<goals>

<goal>exec</goal>

</goals>

</execution>

</executions>

<configuration>

<executable>java</executable>

<includeProjectDependencies>true</includeProjectDependencies>

<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();

        }

    }

}