了解并发进程

Question

我正在构建一个执行特定 IO 绑定任务的脚本。我需要它来尝试下载大量数据集，在丢弃数据本身之前记录有关其大小的某些信息。

问题是我从中获取此数据的源不提供content-length 标头，因此无法事先知道文件大小。这要求我找到一种方法来监控下载过程需要多长时间，并且有一种方法可以终止该过程并继续其他此类进程，以防它花费太长时间（例如，超过 60 秒）。这是必要的，以避免“卡在”非常大的数据集上。

requests 不提供此内置功能，在花费大量时间寻找解决方案后，我决定通过pebble 库运行具有超时的并发进程。我的理解是，这是标准库 multiprocessing 模块的一个小扩展，它添加了一些安全功能，即错误处理和超时（这是我想要的）。

基于Process pool 示例，这是我的代码：

try:
    with ProcessPool(max_workers=4) as pool:
        iterator = pool.map(get_data, process_tuples[3:6], timeout=10)

        while True:
            try:
                rows, cols, filesize, i = next(iterator)
                datasets[i]['rows'] = rows
                datasets[i]['columns'] = cols
                datasets[i]['filesize'] = filesize
            except TimeoutError as error:
                print("Function took longer than %d seconds. Skipping responsible endpoint..." % error.args[1])
            except StopIteration:
                break
finally:
    with open("../../../data/" + FILE_SLUG + "/glossaries/geospatial.json", "w") as fp:
        json.dump(datasets, fp, indent=4)

但这在两个方面与预期行为不同：

1. 我曾认为timeout=10 限制了每个单独的下载过程（由get_data 完成）将花费的时间。但是，当我在一个大文件上运行它时，我收到一个TimeoutError，它指出我的过程花费了 30 多秒。 30 是我输入长度的 3 倍；这根本不是我想要的。那里发生了什么？
2. 当引发TimeoutError 时，进程不会放弃该运行并移至下一个（我想要的），而是跳转到finally 块（我不想要的）。我认为这是对我的第一个问题的回答的结果。

Answer 1

其实在requests中你可以设置stream=True并使用Response.iter_content()来进一步控制工作流程。

在您的情况下，我们可以在下载/迭代响应数据时跟踪经过的时间：

import time
import requests

def get_content(url, timeout):
    """
    Get response data from url before timeout
    """
    start = time.time()
    data = ''
    response = requests.get(url, stream=True)

    for chunk in response.iter_content(chunk_size = 1024): # You can set a bigger chunk_size for less iterations
        if (time.time() - start) > timeout:
            response.close()
            return {'TimedOut': True, 'data': None}
        else:
            data += chunk

    response.close()
    return {'TimedOut': False, 'data': data}

所以基本上你设置了一个timeout的值，如果数据太大或者网络太慢，一旦花费超过timeout就会返回一个结果，那些不完整的数据将被垃圾回收。

接下来因为是IO密集型任务，我们可以使用threading或者multiprocessing来完成任务，这里以threading为例

import threading, Queue

def worker(queue):
    while not queue.empty():
        url = queue.get()

        result = get_content(url, 60)

        # Do other stuff

if __name__ == '__main__':
    limit = 10 # number of threads to use
    thread_pool = [None] * limit
    queue = Queue.Queue()
    urls = ['xxxx', 'xxxxx']

    for url in urls:
        queue.put(url)

    for thread in thread_pool:
        thread = threading.Thread(target=worker, args=(queue, ))
        thread.start()