3

我需要一些帮助来理解为什么合并两个流会阻塞 class 的一个 spout FixedBatchSpout。

简短描述：我正在尝试合并两个流 s1 和 s2，但调用topology.merge(s1, s2)阻止了FixedBatchSpouts1 起源的（三叉戟喷口），而BaseRichSpout来自 s2 的（风暴喷口）似乎工作正常。

详细信息：在下面的主要方法中，只需添加该行即可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);
    }
}

感谢您的帮助，谢谢！雅克

4

0 回答 0