一、概念

Apache Storm作為流數(shù)據(jù)的實(shí)時(shí)處理框架，官網(wǎng)給出了如下模型：

圖中“水龍頭”便是spout [spa?t] 出水管，“閃電”便是bolt [b??lt]，“箭頭”表達(dá)的是數(shù)據(jù)的流轉(zhuǎn)，“水龍頭”、“閃電”和“箭頭”組成的有向無(wú)環(huán)圖稱(chēng)為T(mén)opology（拓?fù)洌?/p>

使用Storm框架進(jìn)行流數(shù)據(jù)的實(shí)時(shí)處理，就需要編寫(xiě)“水龍頭”和“閃電”的處理邏輯，并將它們通過(guò)Topology串接在一起，構(gòu)建實(shí)時(shí)處理的業(yè)務(wù)邏輯。

具體的做法是：

（1）實(shí)時(shí)數(shù)據(jù)源，如kafka，接入到“水龍頭”Spout中；

（2）Spout讀取源數(shù)據(jù)并不斷地發(fā)出數(shù)據(jù)到后續(xù)Bolt中，這些數(shù)據(jù)稱(chēng)為T(mén)uple（元組）；

（3）Bolt對(duì)發(fā)送過(guò)來(lái)的數(shù)據(jù)Tuple進(jìn)行處理，完成數(shù)據(jù)流轉(zhuǎn)換；

讀到這里，可能還是很迷惑，我們以常見(jiàn)的示例統(tǒng)計(jì)詞頻（heart.txt）來(lái)進(jìn)行說(shuō)明：

Take me to your heart
Take me to your soul
Give me your hand and hold me
Show me what love is
Be my guiding star
It's easy take me to your heart
Standing on a mountain high
Looking at the moon through a clear blue sky

我們可以設(shè)計(jì)一個(gè)topology：

WordSourceSpout：讀取heart.txt，并逐行發(fā)送數(shù)據(jù)流Stream，每行即為一個(gè)Tuple；
WordSplitBolt：拆分Tuple，并將單詞Tuple發(fā)出到下個(gè)Bolt；
WordCountBolt：對(duì)單詞的頻率進(jìn)行累加計(jì)算；

二、編程

1.Topology是如何構(gòu)建的？

Topology是通過(guò)TopologyBuilder來(lái)構(gòu)建的，提供setSpout和setBolt方法來(lái)配置Spout和Bolt，這兩個(gè)方法都具有3個(gè)參數(shù)，比較類(lèi)似，以setSpout為例，第1個(gè)參數(shù)表示Stream的名稱(chēng)，第2個(gè)參數(shù)表示stream的處理對(duì)象，第3個(gè)參數(shù)表示并發(fā)數(shù)，也就是同時(shí)運(yùn)行多少個(gè)任務(wù)來(lái)處理Stream。先來(lái)看一段代碼：

TopologyBuilder topologyBuilder = new TopologyBuilder();
WordSourceSpout spout = new WordSourceSpout();
WordSplitBolt splitBlot = new WordSplitBolt();
WordCountBolt countBlot = new WordCountBolt();
topologyBuilder.setSpout("sentences", spout, 2);
topologyBuilder.setBolt("split",splitBlot , 8).shuffleGrouping("sentences");
topologyBuilder.setBolt("count",countBlot , 8).fieldGrouping("split",new Fields(“word”));

上面定義了兩個(gè)Bolt，它們之間數(shù)據(jù)流的關(guān)聯(lián)關(guān)系：第1個(gè)Bolt聲明其輸出Stream的名稱(chēng)為split，而第2個(gè)Bolt訂閱的Stream為split。countBlot 通過(guò)fieldGroupings()在word上具有相同字段的所有Tuple發(fā)送到同一個(gè)任務(wù)中進(jìn)行統(tǒng)計(jì)。

2.Spout和Bolt是如何定義的？

編程模型中，Spout和Bolt都稱(chēng)為組件Component。

WordSourceSpout 需要繼承BaseRichSpout，其類(lèi)結(jié)構(gòu)關(guān)系為：

BaseRichSpout--繼承--BaseComponent--實(shí)現(xiàn)--IComponent
BaseRichSpout--實(shí)現(xiàn)--IRichSpout--實(shí)現(xiàn)--ISpout

ISpout接口的定義為：

public interface ISpout extends Serializable {
    /**
     * Called when a task for this component is initialized within a worker on the cluster. It provides the spout with the environment in
     * which the spout executes.
     *
     * 
This includes the:
     *
     * @param conf      The Storm configuration for this spout. This is the configuration provided to the topology merged in with cluster
     *                  configuration on this machine.
     * @param context   This object can be used to get information about this task's place within the topology, including the task id and
     *                  component id of this task, input and output information, etc.
     * @param collector The collector is used to emit tuples from this spout. Tuples can be emitted at any time, including the open and
     *                  close methods. The collector is thread-safe and should be saved as an instance variable of this spout object.
     */
    void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector);


    /**
     * Called when an ISpout is going to be shutdown. There is no guarentee that close will be called, because the supervisor kill -9's
     * worker processes on the cluster.
     *
     * 
The one context where close is guaranteed to be called is a topology is killed when running Storm in local mode.
     */
    void close();


    /**
     * Called when a spout has been activated out of a deactivated mode. nextTuple will be called on this spout soon. A spout can become
     * activated after having been deactivated when the topology is manipulated using the `storm` client.
     */
    void activate();


    /**
     * Called when a spout has been deactivated. nextTuple will not be called while a spout is deactivated. The spout may or may not be
     * reactivated in the future.
     */
    void deactivate();


    /**
     * When this method is called, Storm is requesting that the Spout emit tuples to the output collector. This method should be
     * non-blocking, so if the Spout has no tuples to emit, this method should return. nextTuple, ack, and fail are all called in a tight
     * loop in a single thread in the spout task. When there are no tuples to emit, it is courteous to have nextTuple sleep for a short
     * amount of time (like a single millisecond) so as not to waste too much CPU.
     */
    void nextTuple();


    /**
     * Storm has determined that the tuple emitted by this spout with the msgId identifier has been fully processed. Typically, an
     * implementation of this method will take that message off the queue and prevent it from being replayed.
     */
    void ack(Object msgId);


    /**
     * The tuple emitted by this spout with the msgId identifier has failed to be fully processed. Typically, an implementation of this
     * method will put that message back on the queue to be replayed at a later time.
     */
    void fail(Object msgId);
}

WordCountBolt需要繼承BaseBasicBolt，其類(lèi)結(jié)構(gòu)關(guān)系為：

BaseBasicBolt--繼承--BaseComponent--實(shí)現(xiàn)--IBasicBolt--IComponent

IBasicBolt接口的定義為：

public interface IBasicBolt extends IComponent {
    void prepare(Map topoConf, TopologyContext context);


    /**
     * Process the input tuple and optionally emit new tuples based on the input tuple.
     *
     * 
All acking is managed for you. Throw a FailedException if you want to fail the tuple.
     */
    void execute(Tuple input, BasicOutputCollector collector);


    void cleanup();
}

IComponent接口的定義：

/**
 * Common methods for all possible components in a topology. This interface is used when defining topologies using the Java API.
 */
public interface IComponent extends Serializable {


    /**
     * Declare the output schema for all the streams of this topology.
     *
     * @param declarer this is used to declare output stream ids, output fields, and whether or not each output stream is a direct stream
     */
    void declareOutputFields(OutputFieldsDeclarer declarer);


    /**
     * Declare configuration specific to this component. Only a subset of the "topology.*" configs can be overridden. The component
     * configuration can be further overridden when constructing the topology using {@link TopologyBuilder}
     */
    Map<String, Object> getComponentConfiguration();


}

Storm框架基本邏輯為：

Spout組件通過(guò)Open方法進(jìn)行SpoutOutputCollector（Spout輸出收集器）的初始化，Storm調(diào)用Spout組件的nextTuple方法請(qǐng)求tuple時(shí)，便通過(guò)SpoutOutputCollector的emit方法發(fā)送一個(gè)tuple。Bolt組件通過(guò)execute方法接收到tuple，并對(duì)tuple進(jìn)行數(shù)據(jù)處理。