Python 并发 executor.map() 和 submit()

Question

我正在学习如何使用 concurrent withexecutor.map()和executor.submit()。

我有一个包含 20 个 url 的列表，想同时发送 20 个请求，问题是.submit()从一开始就以与给定列表不同的顺序返回结果。我读过它map()可以满足我的需要，但我不知道如何用它编写代码。

下面的代码对我来说很完美。

问题：是否有任何代码块map()等同于下面的代码，或者任何排序方法可以submit()按给定列表的顺序对结果列表进行排序？

import concurrent.futures

import urllib.request

URLS = ['http://www.foxnews.com/',

        'http://www.cnn.com/',

        'http://europe.wsj.com/',

        'http://www.bbc.co.uk/',

        'http://some-made-up-domain.com/']

# Retrieve a single page and report the url and contents

def load_url(url, timeout):

    with urllib.request.urlopen(url, timeout=timeout) as conn:

        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:

    # Start the load operations and mark each future with its URL

    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}

    for future in concurrent.futures.as_completed(future_to_url):

        url = future_to_url[future]

        try:

            data = future.result()

        except Exception as exc:

            print('%r generated an exception: %s' % (url, exc))

        else:

            print('%r page is %d bytes' % (url, len(data)))

Answer 1

这是您现有代码的地图版本。请注意，回调现在接受一个元组作为参数。我在回调中添加了一个 try\except，因此结果不会抛出错误。结果根据输入列表排序。from concurrent.futures import ThreadPoolExecutorimport urllib.requestURLS = ['http://www.foxnews.com/',        'http://www.cnn.com/',        'http://www.wsj.com/',        'http://www.bbc.co.uk/',        'http://some-made-up-domain.com/']# Retrieve a single page and report the url and contentsdef load_url(tt):  # (url,timeout)    url, timeout = tt    try:      with urllib.request.urlopen(url, timeout=timeout) as conn:         return (url, conn.read())    except Exception as ex:        print("Error:", url, ex)        return(url,"")  # error, return empty stringwith ThreadPoolExecutor(max_workers=5) as executor:    results = executor.map(load_url, [(u,60) for u in URLS])  # pass url and timeout as tuple to callback    executor.shutdown(wait=True) # wait for all complete    print("Results:")for r in results:  # ordered results, will throw exception here if not handled in callback    print('   %r page is %d bytes' % (r[0], len(r[1])))输出Error: http://www.wsj.com/ HTTP Error 404: Not FoundResults:   'http://www.foxnews.com/' page is 320028 bytes   'http://www.cnn.com/' page is 1144916 bytes   'http://www.wsj.com/' page is 0 bytes   'http://www.bbc.co.uk/' page is 279418 bytes   'http://some-made-up-domain.com/' page is 64668 bytes

Answer 2

在不使用该方法的情况下，您不仅map可以使用URL 作为值，还可以使用它们在列表中的索引来enumerate构建dict。然后，您可以使用索引作为键从调用返回的对象future_to_url构建一个字典，这样您就可以在字典的长度上迭代索引，以按照与原始列表中相应项目相同的顺序读取字典：futureconcurrent.futures.as_completed(future_to_url)with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:    # Start the load operations and mark each future with its URL    future_to_url = {        executor.submit(load_url, url, 60): (i, url) for i, url in enumerate(URLS)    }    futures = {}    for future in concurrent.futures.as_completed(future_to_url):        i, url = future_to_url[future]        futures[i] = url, future    for i in range(len(futures)):        url, future = futures[i]        try:            data = future.result()        except Exception as exc:            print('%r generated an exception: %s' % (url, exc))        else:            print('%r page is %d bytes' % (url, len(data)))