将 python 队列保存到文件中

Question

我正在使用 Python Queue cclass 来管理在多个工作线程之间共享的任务列表。实际的代码非常庞大，我仍在努力使其完全没有错误。有时，工作线程会崩溃，我必须重新启动整个例程。在此过程中，我丢失了所有已排队的任务。有没有办法将队列保存到文件中，以便每当我重新启动进程时，任务列表都会从该文件中预加载？

乍一看，当我将任务放入队列中时，我应该同时读取和写入文件。然而，这并没有给我queue.task_done()的功能，并且可能不是最优化的解决方案。任何想法将不胜感激。

Answer 1

您是否考虑过酸洗您的队列？

Answer 2

有多种方法可以实现此目的，包括 pickle 模块...

但在我看来，只需逐行写入文件，队列中的每个元素包含其他属性的列就很简单您可能想要保存，例如 task_done。

示例：

element1, True
element2, False
...

在 python 中，读取这样格式的文件非常容易，有点像：

for line in file('path/file.ext'):
    name, state = line.split(sep_char)
    #and them insert into the queue...

Answer 3

执行此操作的简单方法是对消息队列使用 AMQP，并让消息代理为您处理消息。我使用 RabbitMQ 作为具有持久持久队列的消息代理实现了一个类似的系统。当我在虚拟 Linux 服务器上使用过时的 1.72 服务器版本时，这些消息甚至在 RabbitMQ 服务器软件崩溃时幸存下来，该服务器只有 512M RAM 和大约一百万条正在运行的消息。

我这样做的方法是每种类型的工作人员消耗来自不同队列的消息。如果我需要多个该类型的工作人员，则消息队列会自动循环，并且如果工作人员无法完成消息的处理，他们只是不确认该消息并将其返回到队列中。

我编写了一个大约 80 行代码的小垫片模块，放在 kombu 前面，后来重写了它以使用 py-amqplib 。如果我早点知道 haigha 我就会使用它，因为它与 AMQP 规范文档非常匹配。

我不推荐 kombu，因为它的调试非常复杂，并且以奇怪的方式偏离 AMQP 标准。看一下 haigha，因为尽管该文档不过是 PyPi 上的一个示例代码片段，但它比 kombu 或 amqplib 的记录更好，因为您可以使用 AMQP 规范作为 haigha 文档。

Answer 4

我可以提供的一个简单选项是将数据库表包装在一个类中并将其用作队列。自动增量列对此会产生奇迹（下一个要删除的项目是 ID 最低的项目）。

class dbQueue:
  init():
    # Pick some random id for this run (or set it to some thing you know).
  put():
    # Insert entry into table
  get():
    # The update .. select combo removes the need for a database that has transactions.
    # If no entries bear your ID:
      # Update the next entry that is not already marked with your ID.
    # Select the entry that matches your ID and return it.
  task_done():
    # Delete the entry with your ID.

这不会有最好的性能，具体取决于队列更新的频率，即使是内存中的 sqlite 数据库也不会像链表结构那么快。另一方面是您可以使用任何可以访问数据库的工具查看数据库，这样您就可以看到哪个数据库正在进行中。

Answer 5

在worker和master之间实现握手机制。

主节点有一个任务列表，在将它们放入队列之前，将列表腌制到文件中。
然后将任务插入队列。当工作人员完成后，他会发回一条 ACK 消息。
仅在此时取消任务列表并删除相应的 id。