Apache Storm Bolt 无法接收来自其他 Bolt 发射的任何元组答案

【问题标题】：Apache Storm Bolt cannot receive any Tuple from other Bolt emitApache Storm Bolt 无法接收来自其他 Bolt 发射的任何元组
【发布时间】：2018-01-07 10:27:26
【问题描述】：

我是 Storm 的新手。我想使用一个名为 'tileClean' 的螺栓来发射单个 Stream，而其他五个螺栓同时接收 Stream。像这样： flow image

如您所见，“一、二、三、四、五”螺栓将同时收到相同的数据。但实际上，“一、二、三、四、五”螺栓无法接收任何数据。有我的代码：

@Override
public void execute(TupleWindow inputWindow) {
    logger.debug("clean");
    List<Tuple> tuples = inputWindow.get();
    //logger.debug("clean phrase. tuple size is : {}", tuples.size());
    for (Tuple input : tuples) {
        // some other code..

        //this._collector.emit(input, new Values(nal));
        this._collector.emit("stream_id_one", input, new Values(nal));
        this._collector.emit("stream_id_two", input, new Values(nal));
        this._collector.emit("stream_id_three", input, new Values(nal));
        this._collector.emit("stream_id_four", input, new Values(nal));
        this._collector.emit("stream_id_five", input, new Values(nal));

        this._collector.ack(input);
    }
}

@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declare(new Fields(BoltConstant.EMIT_LOGOBJ));
    declarer.declareStream("stream_id_one", new Fields(BoltConstant.EMIT_LOGOBJ));
    declarer.declareStream("stream_id_two", new Fields(BoltConstant.EMIT_LOGOBJ));
    declarer.declareStream("stream_id_three", new Fields(BoltConstant.EMIT_LOGOBJ));
    declarer.declareStream("stream_id_four", new Fields(BoltConstant.EMIT_LOGOBJ));
    declarer.declareStream("stream_id_five", new Fields(BoltConstant.EMIT_LOGOBJ));
}

拓扑集是：

builder.setBolt("tileClean", cleanBolt, 1).shuffleGrouping("assembly");    
builder.setBolt("OneBolt", OneBolt, 1).shuffleGrouping("tileClean", "stream_id_one");
builder.setBolt("TwoBolt", TwoBolt, 1).shuffleGrouping("tileClean", "stream_id_two");
builder.setBolt("ThreeBolt", ThreeBolt, 1).shuffleGrouping("tileClean", "stream_id_three");
builder.setBolt("FourBolt", FourBolt, 1).shuffleGrouping("tileClean", "stream_id_four");
builder.setBolt("FiveBolt", FiveBolt, 1).shuffleGrouping("tileClean", "stream_id_five");

tileClean 可以接收从assymble 发出的元组，但其他螺栓无法接收。

我的代码有什么不正确的吗？

【问题讨论】：

标签： java streaming apache-storm topology bolt

【解决方案1】：

由于您省略了“for 循环”语句和第一个 collector.emit 语句之间的代码，因此消息无法通过的一种可能性是在省略的代码之间进行适当的错误处理。您可以通过在“collector.emit”语句之前记录来确保放置 try-catch 块或调试，以检查您的代码是否确实到达那里。

上面的内容也可以在storm-ui上检查，它会显示在spout/bolts之间传输消息的拓扑指标。它还报告任务执行之间可能发生的任何错误消息。

另一种可能性是，如果您使用的是多节点集群，如果您的任务分散在节点上（即，如果您在拓扑配置中分配了超过 1 个工作人员），请确保机器可以在storm.yaml文件中配置的指定端口上通过网络相互通信。

【讨论】：