Apache Storm：确认不起作用答案

【问题标题】：Apache Storm: Ack not workingApache Storm：确认不起作用
【发布时间】：2015-11-27 02:17:26
【问题描述】：

我正在尝试实现有保证的消息处理，但没有调用 Spout 上的 ack 或 fail 方法。

我正在通过 spout 传递消息 ID 对象。我用每个螺栓传递元组并在每个螺栓中调用collector.ack(tuple)。

问题 ack 或 fail 没有被调用，我不知道为什么？

这是一个简短的代码示例。

使用 BaseRichSpout 的 Spout 代码

public void nextTuple() {
    for( String usage : usageData ) {
    .... further code ....

    String msgID = UUID.randomUUID().toString()
                    + System.currentTimeMillis();

    Values value = new Values(splitUsage[0], splitUsage[1],
                    splitUsage[2], msgID);
    outputCollector.emit(value, msgID);
   }
}

@Override
public void ack(Object msgId) {
    this.pendingTuples.remove(msgId);
    LOG.info("Ack " + msgId);
}

@Override
public void fail(Object msgId) {
    // Re-emit the tuple
    LOG.info("Fail " + msgId);
    this.outputCollector.emit(this.pendingTuples.get(msgId), msgId);
}

使用 BaseRichBolt 的螺栓代码

@Override
public void execute(Tuple inputTuple) {

this.outputCollector.emit(inputTuple, new Values(serverData, msgId));

this.outputCollector.ack(inputTuple);
}

最终螺栓

@Override
public void execute(Tuple inputTuple) {
  ..... Simply reports does not emit .....
  this.outputCollector.ack(inputTuple);

}

【问题讨论】：

标签： java apache-storm

【解决方案1】：

ack 不起作用的原因是在 spout 中使用了 for 循环。将此更改为发射下方的计数器循环版本，它可以工作。

示例

        index++;
        if (index >= dataset.size()) {
            index = 0;
        }

进一步感谢邮件列表信息。这是因为 Spout 在单个线程上运行并且会阻塞在 for 循环中，因为下一个元组不会返回，因此它永远无法调用 ACK 方法。

【讨论】：