一、概念

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

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

使用Storm框架進(jìn)行流數(shù)據(jù)的實(shí)時(shí)處理，就需要編寫“水龍頭”和“閃電”的處理邏輯，并將它們通過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ù)稱為Tuple（元組）；

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

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

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：對單詞的頻率進(jìn)行累加計(jì)算；

二、編程

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

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

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的名稱為split，而第2個(gè)Bolt訂閱的Stream為split。countBlot 通過fieldGroupings()在word上具有相同字段的所有Tuple發(fā)送到同一個(gè)任務(wù)中進(jìn)行統(tǒng)計(jì)。

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

編程模型中，Spout和Bolt都稱為組件Component。

WordSourceSpout 需要繼承BaseRichSpout，其類結(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，其類結(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組件通過Open方法進(jìn)行SpoutOutputCollector（Spout輸出收集器）的初始化，Storm調(diào)用Spout組件的nextTuple方法請求tuple時(shí)，便通過SpoutOutputCollector的emit方法發(fā)送一個(gè)tuple。Bolt組件通過execute方法接收到tuple，并對tuple進(jìn)行數(shù)據(jù)處理。