Storm ackerの混乱と保証されたメッセージ処理

Question

Storm's Guaranteeing Message Processingを学習しており、この部分のいくつかのコンセプトで混乱しています。

スパウトによって放出されたメッセージが完全に処理されることを保証するために、ストームはこれを達成するためにアッカーを使用します。スパウトがタプルを発行するたびに、ackerはタプルツリーのステータスを格納するために0に初期化された "ack val"を割り当てます。このタプルの下流側のボルトが新しいタプルを出力するか、「古い」タプルを出力するたびに、タプルIDは「ack val」とXORになります。 ackerは、タプルが完全に処理されたことを知るために、「ack val」が0であるかどうかをチェックするだけでよい。以下のコードを見てみましょう：

public class WordReader implements IRichSpout { 
    ... ... 
while((str = reader.readLine()) != null){ 
    this.collector.emit(new Values(str), str); 
    ... ... 
} 

上記のコードは、「Getting Started with Storm」チュートリアルの単語カウントプログラムのスパウトです。 emitメソッドでは、2番目のパラメータ "str"がmessageIdです。私はこのパラメータで混乱しています： 1）わかっているように、スパウトやボルトに関係なくタプル（メッセージ）が出されるたびに、そのメッセージに64ビットのmessageIdを割り当てるのはStormの責任でなければなりません。あれは正しいですか？または、ここでは "str"はこのメッセージの人間が読める別名です。 2）1）の答えにかかわらず、ここでは "str"は2つの異なるメッセージで同じ単語になります。なぜなら、テキストファイルには重複する単語がたくさんあるからです。これが正しければ、Stormはどのようにして異なるメッセージを区別しますか？そして、このパラメータの意味は何ですか？

public class RandomIntegerSpout extends BaseRichSpout { 
    private long msgId = 0; 
    collector.emit(new Values(..., ++msgId), msgId); 
} 

これは、私はそれがあるべきだと思うものにずっと近い： 3）いくつかのコードの一部では、私はいくつかの注ぎ口がスパウトにメッセージIDを設定するには、次のコードを使用する方法が発する参照メッセージIDがあるべき異なるメッセージ間で全く異なる。しかし、このコードの場合、もう一つの混乱があります。異なるエグゼクティブの間でプライベートフィールド "msgid"はどうなりますか？各エグゼキュータには独自のmsgIdが0として初期化されているため、異なるエグゼキュータ内のメッセージには0,1,2などの名前が付けられます。そして、Stormはこれらのメッセージをどのように区別しますか？

私はストームの初心者ですので、これらの問題は素朴です。誰かが私を理解するのを助けてくれることを願っている。ありがとう！

Answer 1

メッセージIDについて：内部的には64ビット値である可能性がありますが、この64ビット値はSpout内のemit()で提供されるmsgIDオブジェクトからのハッシュとして計算されます。したがって、任意のオブジェクトをメッセージID（2つのオブジェクトが同じ値にハッシュする確率がゼロに近い確率）として渡すことができます。

使用についてstr：この例では、strには1行（単語ではない）が含まれていて、ドキュメントには同じ行が2回含まれている可能性はほとんどありません（空白行がない場合）。

メッセージIDとしてのカウンタについて：複数のスパウトが並行して実行されている場合は、メッセージIDの競合が発生し、フォールトトレランス機能が損なわれます。

カウンタアプローチを修正する場合は、各カウンタを異なる方法で初期化する必要があります（最高：1...#SpoutTasks）。これにはtaskIDを使用できます（ユニークで、でアクセスできます）。Spout.open()で提供されています。基本的には、すべての並列スパウト・タスクのすべてのtaskIDを取得し、ソートし、各スパウト・タスクにその発注番号を割り当てます。さらに、1ではなく、「並列スパウトの数」を増やす必要があります。