一個最簡單的JStorm例子

最簡單的JStorm例子分為以下幾個步驟：

1、生成Topology

Map conf = new HashMp();
//topology所有自定義的配置均放入這個Map

TopologyBuilder builder = new TopologyBuilder();
//創建topology的生成器

int spoutParal = get("spout.parallel", 1);
//獲取spout的並發設置

SpoutDeclarer spout = builder.setSpout(SequenceTopologyDef.SEQUENCE_SPOUT_NAME,
                new SequenceSpout(), spoutParal);
//創建Spout， 其中new SequenceSpout() 為真正spout對象，SequenceTopologyDef.SEQUENCE_SPOUT_NAME 為spout的名字，注意名字中不要含有空格

int boltParal = get("bolt.parallel", 1);
//獲取bolt的並發設置

BoltDeclarer totalBolt = builder.setBolt(SequenceTopologyDef.TOTAL_BOLT_NAME, new TotalCount(),
                boltParal).shuffleGrouping(SequenceTopologyDef.SEQUENCE_SPOUT_NAME);
//創建bolt， SequenceTopologyDef.TOTAL_BOLT_NAME 為bolt名字，TotalCount 為bolt對象，boltParal為bolt並發數，
//shuffleGrouping（SequenceTopologyDef.SEQUENCE_SPOUT_NAME）， 
//表示接收SequenceTopologyDef.SEQUENCE_SPOUT_NAME的數據，並且以shuffle方式，
//即每個spout隨機輪詢發送tuple到下一級bolt中

int ackerParal = get("acker.parallel", 1);
Config.setNumAckers(conf, ackerParal);
//設置表示acker的並發數

int workerNum = get("worker.num", 10);
conf.put(Config.TOPOLOGY_WORKERS, workerNum);
//表示整個topology將使用幾個worker

conf.put(Config.STORM_CLUSTER_MODE, "distributed");
//設置topolog模式為分布式，這樣topology就可以放到JStorm集群上運行

StormSubmitter.submitTopology(streamName, conf,
                builder.createTopology());
//提交topology

2、IRichSpout

IRichSpout 為最簡單的Spout接口

 IRichSpout{

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
    }

    @Override
    public void close() {
    }

    @Override
    public void activate() {
    }

    @Override
    public void deactivate() {
    }

    @Override
    public void nextTuple() {
    }

    @Override
    public void ack(Object msgId) {
    }

    @Override
    public void fail(Object msgId) {
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

 其中注意：

• spout對象必須是繼承Serializable， 因此要求spout內所有數據結構必須是可序列化的
• spout可以有構造函數，但構造函數只執行一次，是在提交任務時，創建spout對象，因此在task分配到具體worker之前的初始化工作可以在此處完成，一旦完成，初始化的內容將攜帶到每一個task內（因為提交任務時將spout序列化到文件中去，在worker起來時再將spout從文件中反序列化出來）。
• open是當task起來后執行的初始化動作
• close是當task被shutdown后執行的動作
• activate 是當task被激活時，觸發的動作
• deactivate 是task被deactive時，觸發的動作
• nextTuple 是spout實現核心， nextuple完成自己的邏輯，即每一次取消息后，用collector 將消息emit出去。
• ack， 當spout收到一條ack消息時，觸發的動作，詳情可以參考 ack機制
• fail， 當spout收到一條fail消息時，觸發的動作，詳情可以參考 ack機制
• declareOutputFields， 定義spout發送數據，每個字段的含義
• getComponentConfiguration 獲取本spout的component 配置

3、Bolt

IRichBolt {

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
    }

    @Override
    public void execute(Tuple input) {
    }

    @Override
    public void cleanup() {
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

}

 其中注意：

• bolt對象必須是繼承Serializable， 因此要求spout內所有數據結構必須是可序列化的
• bolt可以有構造函數，但構造函數只執行一次，是在提交任務時，創建bolt對象，因此在task分配到具體worker之前的初始化工作可以在此處完成，一旦完成，初始化的內容將攜帶到每一個task內（因為提交任務時將bolt序列化到文件中去，在worker起來時再將bolt從文件中反序列化出來）。
• prepare是當task起來后執行的初始化動作
• cleanup是當task被shutdown后執行的動作
• execute是bolt實現核心， 完成自己的邏輯，即接受每一次取消息后，處理完，有可能用collector 將產生的新消息emit出去。 ** 在executor中，當程序處理一條消息時，需要執行collector.ack， 詳情可以參考 ack機制 ** 在executor中，當程序無法處理一條消息時或出錯時，需要執行collector.fail ，詳情可以參考 ack機制
• declareOutputFields， 定義bolt發送數據，每個字段的含義
• getComponentConfiguration 獲取本bolt的component 配置

4、編譯

在Maven中配置

        <dependency>
            <groupId>com.alibaba.jstorm</groupId>
            <artifactId>jstorm-client</artifactId>
            <version>0.9.3.1</version>
            <scope>provided</scope>
        </dependency> 


         <dependency>
            <groupId>com.alibaba.jstorm</groupId>
            <artifactId>jstorm-client-extension</artifactId>
            <version>0.9.3.1</version>
            <scope>provided</scope>
        </dependency>

如果找不到jstorm-client和jstorm-client-extension包，可以自己下載jstorm源碼進行編譯，請參考 源碼編譯

打包時，需要將所有依賴打入到一個包中

<build>
        <plugins>

            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <archive>
                        <manifest>
                            <mainClass>storm.starter.SequenceTopology</mainClass>
                        </manifest>
                    </archive>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.6</source>
                    <target>1.6</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

5、提交jar

jstorm jar xxxxxx.jar com.alibaba.xxxx.xx parameter

• xxxx.jar 為打包后的jar
• com.alibaba.xxxx.xx 為入口類，即提交任務的類
• parameter即為提交參數