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

Question

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

我想确保 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 个 fetcher，这需要几秒到半分钟，我需要补偿。

3) 我想确定来自 Queue.Get() 的数据是由一个没有超时的 Fetcher 放在那里的(因为从理论上讲，一个 fetcher 可能将Queue中的数据，超时的时候，被拍死了。。。)那个数据应该dump掉。

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

Answer 1

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

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

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

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

with fetcherLock:
    fetcherProcess.terminate()

这样，您的队列就不会因为在队列访问期间终止提取程序而受到破坏。

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