python - 使用 ZooKeeper 管理正在处理或已处理的任务

Question

我有一个 python 脚本，它定期扫描目录、处理新文件。每个文件都需要很长时间来处理(许多小时)。我目前在一台计算机上运行脚本，将已处理文件的名称写入本地文件。不花哨或健壮，但它或多或少有效。我想使用多台工作机器来提高吞吐量(和稳健性)。我的目标是让它尽可能简单。一个 zookeeper 集群很容易获得。

我的计划是在 zookeeper 中有一个目录“started_files”，其中包含带有文件名的临时节点，已知文件名是唯一的。我会有另一个目录“completed_files”，其中包含带有文件名的常规节点。在伪代码中，

if filename does not exist in completed files:
    try:
        create emphemeral node filename in started files
        process(filename)
        create node filename in completed files
    except node exists error:
        do nothing, another worker is processing it

我的第一个问题是这是否安全。在任何情况下，两台不同的机器都可以成功创建同一个节点吗？我不完全理解 doc .将一个文件处理两次不会造成任何糟糕的结果，但我更希望它在原则上是正确的。

其次，这是一个不错的方法吗？还有另一种明显更好的方法吗？我每天要处理 10 个文件，所以这部分应用程序的性能对我来说并不重要(我当然希望处理文件的速度更快)。或者，我可以使用另一个只有一个实例(或选择一个领导者)的脚本来扫描文件并将它们放入队列中。我可以修改导致这些文件神奇地出现在第一位的代码。我可以用 celery 或 Storm 。然而，所有这些替代方案都扩大了我试图保持小而简单的范围。

Answer 1

一般来说，您的方法应该可行。有可能，您将 znode 写入 ZooKeeper 的方式配置为连续创建相同路径(如果存在)将失败。

对于临时 znode，您已经很清楚地发现，如果客户端关闭与 ZooKeeper 的连接，它们会自动消失，这在计算节点出现故障的情况下特别有用。

其他节点实际上可以使用临时 znode 监视路径，以便确定何时扫描新任务是个好主意。

甚至可以在 ZooKeeper 之上实现一个队列，例如使用 znode 的排序；可能有更好的方法。

总的来说，我认为具有发布订阅模式的消息队列系统的扩展性会更好一些。在那种情况下，您只需要考虑如何重新安排发生故障的计算节点的作业。