JStorm环境搭建

package storm.cookbook;
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;
/**
 * Author: ashrith
 * Date: 8/26/13
 * Time: 12:03 PM
 * Desc: setup the topology and submit it to either a local of remote Storm cluster depending on the arguments
 *       passed to the main method.
 */
public class HelloWorldTopology {
    /*
     * main class in which to define the topology and a LocalCluster object (enables you to test and debug the
     * topology locally). In conjunction with the Config object, LocalCluster allows you to try out different
     * cluster configurations.
     *
     * Create a topology using ‘TopologyBuilder‘ (which will tell storm how the nodes area arranged and how they
     * exchange data)
     * The spout and the bolts are connected using ‘ShuffleGroupings‘
     *
     * Create a ‘Config‘ object containing the topology configuration, which is merged with the cluster configuration
     * at runtime and sent to all nodes with the prepare method
     *
     * Create and run the topology using ‘createTopology‘ and ‘submitTopology‘
     */
    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("randomHelloWorld", new HelloWorldSpout(), 10);
        builder.setBolt("HelloWorldBolt", new HelloWorldBolt(), 1).shuffleGrouping("randomHelloWorld");
        Config conf = new Config();
        conf.put(Config.NIMBUS_HOST, "localhost");
        conf.put(Config.NIMBUS_THRIFT_PORT, 6627);
        conf.setDebug(true);
        if(args!=null && args.length > 0) {
            conf.setNumWorkers(3);
            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
        } else {
        }
    }
}

package storm.cookbook;
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;
import java.util.Map;
import java.util.Random;
/**
 * Author: ashrith
 * Date: 8/21/13
 * Time: 8:33 PM
 * Desc: spout essentially emits a stream containing 1 of 2 sentences ‘Other Random Word‘ or ‘Hello World‘ based on
 *       random probability. It works by generating a random number upon construction and then generating subsequent
 *       random numbers to test against the original member variable‘s value. When it matches "Hello World" is emitted,
 *       during the remaining executions the other sentence is emitted.
 */
public class HelloWorldSpout extends BaseRichSpout{
    private SpoutOutputCollector collector;
    private int referenceRandom;
    private static final int MAX_RANDOM = 10;
    public HelloWorldSpout() {
        final Random rand = new Random();
        referenceRandom = rand.nextInt(MAX_RANDOM);
    }
    /*
     * declareOutputFields() => you need to tell the Storm cluster which fields this Spout emits within the
     *  declareOutputFields method.
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("sentence"));
    }
    /*
     * open() => The first method called in any spout is ‘open‘
     *           TopologyContext => contains all our topology data
     *           SpoutOutputCollector => enables us to emit the data that will be processed by the bolts
     *           conf => created in the topology definition
     */
    @Override
    public void open(Map conf, TopologyContext topologyContext, SpoutOutputCollector collector) {
        this.collector = collector;
    }
    /*
     * nextTuple() => Storm cluster will repeatedly call the nextTuple method which will do all the work of the spout.
     *  nextTuple() must release the control of the thread when there is no work to do so that the other methods have
     *  a chance to be called.
     */
    @Override
    public void nextTuple() {
        final Random rand = new Random();
        int instanceRandom = rand.nextInt(MAX_RANDOM);
        if(instanceRandom == referenceRandom){
            collector.emit(new Values("Hello World"));
        } else {
            collector.emit(new Values("Other Random Word"));
        }
    }
}

package storm.cookbook;
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;
import java.util.Map;
/**
 * Author: ashrith
 * Date: 8/26/13
 * Time: 11:48 AM
 * Desc: This bolt will consume the produced Tuples from HelloWorldSpout and implement the required counting logic
 */
public class HelloWorldBolt extends BaseRichBolt {
    private int myCount = 0;
    /*
     * prepare() => on create
     */
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
    }
    /*
     * execute() => most important method in the bolt is execute(Tuple input), which is called once per tuple received
     *  the bolt may emit several tuples for each tuple received
     */
    @Override
    public void execute(Tuple tuple) {
        String test = tuple.getStringByField("sentence");
        if(test == "Hello World"){
            myCount++;
            System.out.println("Found a Hello World! My Count is now: " + Integer.toString(myCount));
        }
    }
    /*
     * declareOutputFields => This bolt emits nothing hence no body for declareOutputFields()
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    }
}