Как использовать распараллеливание в понимании set / list, используя asyncio?

Question

Я хочу создать многопроцессорное понимание в 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?

Answer 1

requests не поддерживает asyncio. Если вы хотите использовать истинное асинхронное выполнение, вам нужно посмотреть на библиотеки типа aiohttp или asks

Ваш набор должен быть собран до разгрузки задач, так что вы даже не выполняете дубликаты, вместо того, чтобы оптимизировать результат.

С самим requests вы можете вернуться к run_in_executor, который будет выполнять ваши запросы внутри ThreadPoolExecutor , поэтому на самом деле не асинхронный ввод-вывод:

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 секунды. Однако, учитывая тот факт, что каждая из этих трех задач выполняется в течение десяти секунд до истечения времени ожидания на моем компьютере (моя IDE не разрешена через прокси-сервер), общее время выполнения в двенадцать секунд выглядит вполне параллельным.

Answer 2

asyncio сам по себе не запускает различные функции async одновременно. Однако с помощью Pool.map модуля multiprocessing можно запланировать запуск функций в другом процессе:

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]