apache-storm - トライデント ストリームをマージするとトライデントのスパウトがブロックされますが、ストーム スパウトは機能し続けます

Question

2 つのストリームをマージすると class のスパウトの 1 つがブロックされる理由を理解する助けが必要FixedBatchSpoutです。

簡単な説明: 2 つのストリーム s1 と s2 をマージしようとしていますが、s2からの(嵐の噴出口) が正常に機能しているように見えますが、s1 の元topology.merge(s1, s2)の (トライデントの噴出口) を呼び出すとブロックされます。FixedBatchSpoutBaseRichSpout

詳細: 以下の main メソッドでは、行を追加するだけで、が最初のバッチを過ぎて出力されるtopology.merge(s1, s2);のを防ぎます。FixedBatchSpoutこれmultireduceも同様に起こります。

FixedBatchSpout spout1 = new FixedBatchSpout(new Fields("sentence"), 2,
                new Values("the cow jumped over the moon"),
                new Values("the man went to the store and bought some candy”));

FixedLoopSpout spout2 = new FixedLoopSpout(new Fields("sentence"),
                new Values("THE COW JUMPED OVER THE MOON"),
                new Values("THE MAN WENT TO THE STORE AND BOUGHT SOME CANDY"));

Stream s1 = topology.newStream("hello", spout1);
Stream s2 = topology.newStream("world", spout2);
topology.merge(s1, s2);

public class FixedLoopSpout extends BaseRichSpout {

    Values[] values;
    List<Values> loop = new LinkedList<Values>();
    Iterator<Values> head;
    private SpoutOutputCollector collector;
    private final Fields outputFields;

    private long emitted = 0;

    public FixedLoopSpout(Fields outputFields, Values... values) {
        this.outputFields = outputFields;
        this.values = values;
    }

    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
        for (Values value: this.values) {
            this.loop.add(value);
        }
        this.head = this.loop.iterator();
    }

    public void nextTuple() {
        if (!this.head.hasNext()) {
            // wrap
            this.head = this.loop.iterator();
        }
        this.collector.emit(this.head.next(), this.emitted++);
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(this.outputFields);
    }
}

助けていただければ幸いです、ありがとう！ジャック