如何使用 asyncio 在集合/列表理解中使用并行化？答案

【问题标题】：How to use parallelization in set/list comprehension using asyncio?如何使用 asyncio 在集合/列表理解中使用并行化？
【发布时间】：2018-09-18 05:22:26
【问题描述】：

我想在 Python 3.7 中创建一个多进程理解。

这是我的代码：

async def _url_exists(url):
  """Check whether a url is reachable"""
  request = requests.get(url)
  return request.status_code == 200:

async def _remove_unexisting_urls(rows):
  return {row for row in rows if await _url_exists(row[0])}

rows = [
  'http://example.com/',
  'http://example.org/',
  'http://foo.org/',
]
rows = asyncio.run(_remove_unexisting_urls(rows))

在此代码示例中，我想从列表中删除不存在的 URL。（请注意，我使用的是集合而不是列表，因为我还想删除重复项）。

我的问题是我仍然看到执行是顺序的。 HTTP 请求使执行等待。与串行执行相比，执行时间是相同的。

我做错了吗？
这些 await/async 关键字应该如何与 python 理解一起使用？

【问题讨论】：

标签： python parallel-processing list-comprehension python-asyncio set-comprehension

【解决方案1】：

asyncio 本身不会同时运行不同的async 函数。但是，使用multiprocessing 模块的Pool.map，您可以安排函数在另一个进程中运行：

from multiprocessing.pool import Pool

pool = Pool()

def fetch(url):
    request = requests.get(url)
    return request.status_code == 200

rows = [
  'http://example.com/',
  'http://example.org/',
  'http://foo.org/',
]
rows = [r for r in pool.map(fetch, rows) if r]

【讨论】：

map是Pool的实例方法；你不能在类上调用它（没有传递Pool 的实例给它），所以它需要是Pool.map(Pool(), fetch, set(rows))。虽然with Pool() as pool: rows = [r for r in pool.map(fetch, set(rows)) is r] 可能是更常见的选择。（使用 set(rows) 根据 OP 的关注删除重复项）
糟糕！你是对的，我已经更新了答案。谢谢！

【解决方案2】：

requests 不支持asyncio。如果你想实现真正的异步执行，你将不得不查看像 aiohttp 或 asks 这样的库

您的集合应该在卸载到任务之前构建，因此您甚至不需要执行重复项，而是简化结果。

使用requests 本身，您可以回退到run_in_executor，它将在ThreadPoolExecutor 内执行您的请求，因此不是真正的异步I/O：

import asyncio
import time
from requests import exceptions, get

def _url_exists(url):
    try:
        r = get(url, timeout=10)
    except (exceptions.ConnectionError, exceptions.ConnectTimeout):
        return False
    else:
        return r.status_code is 200

async def _remove_unexisting_urls(l, r):
    # making a set from the list before passing it to the futures
    # so we just have three tasks instead of nine
    futures = [l.run_in_executor(None, _url_exists, url) for url in set(r)]
    return [await f for f in futures]

rows = [ # added some dupes
    'http://example.com/',
    'http://example.com/',
    'http://example.com/',
    'http://example.org/',
    'http://example.org/',
    'http://example.org/',
    'http://foo.org/',
    'http://foo.org/',
    'http://foo.org/',
]

loop = asyncio.get_event_loop()
print(time.time())
result = loop.run_until_complete(_remove_unexisting_urls(loop, rows))
print(time.time())
print(result)

输出

1537266974.403686
1537266986.6789136
[False, False, False]

如您所见，初始化线程池会产生损失，在这种情况下约为 2.3 秒。但是，考虑到这三个任务中的每一个都运行了 10 秒直到我的机器超时（我的 IDE 不允许通过代理），所以总体上 12 秒的执行时间看起来相当并发。

【讨论】：