java - Storm 观察服务有什么吗？

Question

我有一个 bolt ，输入文件不断更新。但我无法获取更新的内容，因为我正在从 prepare() 方法读取文件。我想在不停止或终止拓扑的情况下获取更新的文件。 Storm中有没有类似 watch 服务的东西可以做到这一点？或者有什么不同的方法吗？

Answer 1

解决问题的一种方法是定义一个 Spout，它会定期检查文件是否发生更改。一旦发生，它会发送一个元组，通知您的 bolt 发生了变化。 bolt 将依次重新加载文件。以下是有关实现的一些提示:

拓扑将包含新的监控 spout。您的 Bolt 将订阅它的流以及它需要的任何其他流(Bolt 可以消耗多个流):

topologyBuilder.setSpout("file_checking_spout", new FileCheckingSpout(myMonitoredFile));
topologyBuilder.setBolt("my_bolt", new MyBolt())
    .shuffleGrouping("file_checking_spout")
    .shuffleGrouping("whatever other grouping you need");

Spout 将进行监控。如果只有一个文件要监视，您可以只发出空元组作为通知:

public class FileCheckingSpout extends BaseRichSpout {
    @Override
    public void nextTuple() {
        Thread.sleep(500);
        if (fileChanged()) { // check e.g. file modified timestamp
            collector.emit(new Values());
        }
    }

    @Override 
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields());
    }
    // ...
}

您的 Bolt 现在必须接受有关文件重新加载的通知。它可以区分通知元组，例如使用tuple.getSourceComponent():

class MyBolt implements IRichBolt {
    @Override 
    public void execute(Tuple tuple) {
        if ("file_checking_spout".equals(tuple.getSourceComponent())) {
             reloadFile();
             return;
        }
        // normal processing
    }
    //...
}

您还可以简单地检查 Bolt 的 nextTuple() 中的文件是否发生更改。上面描述的方式更像是“Storm 方式”，因为它分离了关注点并且重新加载不依赖于任何其他流。

PS:当然，只要文件可从 spout 和 Bolt 访问，这就会起作用，即，如果您在集群中运行，则它应该位于共享文件系统上。