使用 Process.Terminate() 时如何解决队列损坏问题

Question

我正在构建一个 Python 脚本/应用程序，它会启动多个所谓的 Fetchers。 他们反过来做一些事情并将数据返回到队列中。

我想确保 Fetcher 的运行时间不超过 60 秒（因为整个应用程序在一小时内运行多次）。

阅读 Python 文档后，我注意到他们说在使用 Process.Terminate() 时要小心，因为它会破坏队列。

我当前的代码：

# Result Queue
resultQueue = Queue();

# Create Fetcher Instance
fetcher = fetcherClass()

# Create Fetcher Process List
fetcherProcesses = []

# Run Fetchers
for config in configList:
    # Create Process to encapsulate Fetcher
    log.debug("Creating Fetcher for Target: %s" % config['object_name'])
    fetcherProcess = Process(target=fetcher.Run, args=(config,resultQueue))

    log.debug("Starting Fetcher for Target: %s" % config['object_name'])
    fetcherProcess.start()
    fetcherProcesses.append((config, fetcherProcess))

# Wait for all Workers to complete
for config, fetcherProcess in fetcherProcesses:
    log.debug("Waiting for Thread to complete (%s)." % str(config['object_name']))
    fetcherProcess.join(DEFAULT_FETCHER_TIMEOUT)
    if fetcherProcess.is_alive():
        log.critical("Fetcher thread for object %s Timed Out! Terminating..." % config['object_name'])
        fetcherProcess.terminate()

# Loop thru results, and save them in RRD
while not resultQueue.empty():
    config, fetcherResult = resultQueue.get()
    result = storage.Save(config, fetcherResult)

我想确保当我的一个 Fetcher 超时时我的队列不会损坏。

最好的方法是什么？

编辑：回应与 sebdelsol 的聊天，澄清几点：

1) 我想尽快开始处理数据，否则我必须同时执行大量磁盘密集型操作。所以为 X_Timeout 休眠主线程不是一种选择。

2）我只需要等待一次超时，但是每个进程，所以如果主线程启动 50 个 fetchers，这需要几秒钟到半分钟，我需要补偿。

3) 我想让 确保 来自 Queue.Get() 的数据由未超时的 Fetcher 放在那里（因为理论上可能是 fetcher 将队列中的数据，当超时发生时，它被枪杀......）应该转储该数据。

发生超时并不是一件很糟糕的事情，这不是一个理想的情况，但损坏的数据更糟糕。

Answer 1

您可以将新的multiprocessing.Lock() 传递给您启动的每个提取器。

在 fetcher 的过程中，一定要用这个锁包裹Queue.put()：

with self.lock:
    self.queue.put(result)

当你需要终止一个 fetcher 的进程时，使用它的锁：

with fetcherLock:
    fetcherProcess.terminate()

这样，您的队列不会因在队列访问期间杀死提取器而损坏。

某些 fetcher 的锁可能会损坏。但是，这不是问题，因为您启动的每个新 fetcher 都有一个全新的锁。

Answer 2

为什么不

1. 创建一个新队列并启动所有将使用该队列的 fetcher 排队。
2. 让您的脚本实际休眠您希望提取器进程获得结果的时间。
3. 从 resultQueue 中获取所有内容 - 它不会损坏，因为您不必终止任何进程。
4. 最后，终止所有仍处于活动状态的 fetcher 进程。
5. 循环！