执行大量HTTP请求，每次异步执行N个。

huangapple
go评论98阅读模式
英文:

Perform large numbers of HTTP requests asyncronously N at a time

问题

以下是翻译好的部分：

我需要执行一项一次性任务，需要遍历数据库中的所有记录，其中有数百万条记录，读取单元格中的值，发出HTTP请求并更新另一个当前为NULL的单元格。

我想通过asyncio以异步方式将它们分批发送，但不要一次发送太多，因为远程服务器可能会封禁我：每秒不超过50个请求或同时进行。

我找到了这段代码：

  1. import asyncio
  2. import aiohttp
  4. async def one(session, url):
  5.     # 请求URL并读取完整或取消
  6.     async with session.get(url) as resp:
  7.         await resp.text()
  9. async def fire(urls):
  10.     loop = asyncio.get_event_loop()
  11.     async with aiohttp.ClientSession() as session:
  12.         tasks = []
  13.         for url in urls:
  14.             tasks.append(loop.create_task(one(session, url)))
  16.         # 10秒超时
  17.         try:
  18.             await asyncio.wait_for(asyncio.gather(*tasks), timeout=10)
  19.         except asyncio.TimeoutError:
  20.             pass
  22. loop = asyncio.get_event_loop()
  23. loop.run_until_complete(fire([urls...]))

但它会一次发送所有请求。

我该如何以每次发送N个请求的方式来做？ 意思是发送N个请求，然后等待1...甚至全部返回值，然后再发送另一批N...以此类推。

英文:

For some one-time task I need to go through all the records in a database of which there are a few millions, read a value in a cell, make a HTTP request and update another cell which is currently NULL.

I want to send all of them by portions, asynchronously, via asyncio. And not too many at a time because remote server may ban me: No more than 50 requests/second or at a time.

I've found this code:

  1. import asyncio
  2. import aiohttp
  5. async def one(session, url):
  6.     # request the URL and read it until complete or canceled
  7.     async with session.get(url) as resp:
  8.         await resp.text()
  11. async def fire(urls):
  12.     loop = asyncio.get_event_loop()
  13.     async with aiohttp.ClientSession() as session:
  14.         tasks = []
  15.         for url in urls:
  16.             tasks.append(loop.create_task(one(session, url)))
  18.         # 10 seconds
  19.         try:
  20.             await asyncio.wait_for(asyncio.gather(*tasks), timeout=10)
  21.         except asyncio.TimeoutError:
  22.             pass
  25. loop = asyncio.get_event_loop()
  26. loop.run_until_complete(fire([urls...]))

But it will send all the request at once.

How could I do it N at a time? Meaning, send N, then wait for 1 ... a few or even all of them to return values, then send another lot of N... and so on.

答案1

得分: 1

以下是代码部分的翻译：

  1. ## 选项 A：仅使用 `asyncio` 批处理
  3. #### Python `<3.11`
  5. ```python
  6. from asyncio import create_task, gather, run, sleep
  8. from aiohttp import ClientSession
  10. async def get_one(session: ClientSession, url: str) -> None:
  11.     print("请求中", url)
  12.     async with session.get(url) as resp:
  13.         text = await resp.text()
  14.         await sleep(2)  # 仅供演示
  15.         print("来自", url, "的响应", text.strip().split("\n", 1)[0])
  17. async def get_all(urls: list[str], num_concurrent: int) -> None:
  18.     url_iterator = iter(urls)
  19.     keep_going = True
  20.     async with ClientSession() as session:
  21.         while keep_going:
  22.             tasks = []
  23.             for _ in range(num_concurrent):
  24.                 try:
  25.                     url = next(url_iterator)
  26.                 except StopIteration:
  27.                     keep_going = False
  28.                     break
  29.                 new_task = create_task(get_one(session, url))
  30.                 tasks.append(new_task)
  31.             await gather(*tasks)
  33. async def main() -> None:
  34.     urls = [
  35.         "https://github.com",
  36.         "https://stackoverflow.com",
  37.         "https://python.org",
  38.     ]
  39.     await get_all(urls, 2)
  41. run(main())

输出：

  1. 请求中 https://github.com
  2. 请求中 https://stackoverflow.com
  3. 来自 https://github.com 的响应 <!DOCTYPE html>
  4. 来自 https://stackoverflow.com 的响应 <!DOCTYPE html>
  5. 请求中 https://python.org
  6. 来自 https://python.org 的响应 <!doctype html>

你会注意到第三个请求（到 python.org）仅在前两个请求都返回响应后才发送。这个设置实际上会批量执行总请求数，每次执行 num_concurrent 个请求。

Python >=3.11

使用较新的 TaskGroup 类，我们可以使 get_all 函数更加简洁：

  1. from asyncio import TaskGroup, run, sleep
  3. from aiohttp import ClientSession
  6. async def get_one(session: ClientSession, url: str) -> None:
  7.     ...  # 与上面相同
  10. async def get_all(urls: list[str], num_concurrent: int) -> None:
  11.     url_iterator = iter(urls)
  12.     keep_going = True
  13.     async with ClientSession() as session:
  14.         while keep_going:
  15.             with TaskGroup() as tg:
  16.                 for _ in range(num_concurrent):
  17.                     try:
  18.                         url = next(url_iterator)
  19.                     except StopIteration:
  20.                         keep_going = False
  21.                         break
  22.                     tg.create_task(get_one(session, url))
  24. ...

选项 B：仅使用 asyncio 中的 Queue

asyncio.Queue 允许我们为其设置最大大小。这样可以限制最大并发执行任务的数量，但我们需要使用 消费者-生产者模式：

  1. from asyncio import Queue, create_task, gather, run, sleep
  3. from aiohttp import ClientSession
  5. async def get_one(session: ClientSession, url: str) -> None:
  6.     ...  # 与上面相同
  8. STOP_SENTINEL = object()
  10. async def consumer(session: ClientSession, q: Queue[str]) -> None:
  11.     url = await q.get()
  12.     while url is not STOP_SENTINEL:
  13.         await get_one(session, url)
  14.         q.task_done()
  15.         url = await q.get()
  16.     q.task_done()
  18. async def main() -> None:
  19.     urls = [
  20.         "https://github.com",
  21.         "https://stackoverflow.com",
  22.         "https://python.org",
  23.     ]
  24.     num_concurrent = 2
  25.     q = Queue(maxsize=num_concurrent)
  26.     async with ClientSession() as session:
  27.         consumers = [
  28.             create_task(consumer(session, q))
  29.             for _ in range(num_concurrent)
  30.         ]
  31.         for url in urls:
  32.             await q.put(url)
  33.         for _ in range(num_concurrent):
  34.             await q.put(STOP_SENTINEL)
  35.         await gather(*consumers)
  37. run(main())

输出：

  1. 请求中 https://github.com
  2. 请求中 https://stackoverflow.com
  3. 来自 https://github.com 的响应 <!DOCTYPE html>
  4. 请求中 https://python.org
  5. 来自 https://stackoverflow.com 的响应 <!DOCTYPE html>
  6. 来自 https://python.org 的响应 <!doctype html>

现在你可以看到，第三个请求可以在前两个返回响应后尽早发送。这可能更有效，尽管设置稍微繁琐一些。

选项 C：使用额外的包

我曾遇到类似的问题，需要在大量的 实际 任务上设置固定数量的 asyncio 任务。为了更容易解决这个问题，我编写了 asyncio-taskpool 包。使用它，我们可以像这样做：

  1. from asyncio import run, sleep
  3. from aiohttp import ClientSession
  4. from asyncio_taskpool import TaskPool
  6. async def get_one(session: ClientSession, url: str) -> None:
  7.     ...  # 与上面相同
  9. async def get_all(urls: list[str], num_concurrent: int) -> None:
  10.     pool = TaskPool()
  11.     async with ClientSession() as session:
  12.         pool.starmap(
  13.             get_one,
  14.             ((session, url) for url in urls),
  15.             num_concurrent=num_concurrent,
  16.         )
  17.         await pool.gather_and_close()
  19. async def main() -> None:
  20.     urls = [
  21.         "https://github.com",
  22.         "https://stackoverflow.com",
  23.         "https://python.org",
  24.     ]
  25.     await get_all(urls, 2)
  27. run(main())

输出：（与 Queue 方法相同）

  1. 请求中 https://github.com
  2. 请求中 https://stackoverflow.com
  3. 来自 https://github.com 的响应 <!DOCTYPE html>
  4. 请求中 https://python.org
  5. 来自 https://stackoverflow.com 的响应 <!DOCTYPE html>
  6. 来自 https://python.org 的响应 <!doctype html>

你会再次注意到，第三个请求仅在其他两个中的至少一个返回响应后才会被发出。你可以尝试使用更多的任务。在任何给定时间内并发执行的数量不会超过传递给 map 的 num_concurrent。

我尽量模拟了标准的 multiprocessing.Pool 接口，这在处理长时间运行的任务时更方便。

英文:

Option A: With just asyncio in batches

Python &lt;3.11

  1. from asyncio import create_task, gather, run, sleep
  3. from aiohttp import ClientSession
  5. async def get_one(session: ClientSession, url: str) -&gt; None:
  6.     print(&quot;Requesting&quot;, url)
  7.     async with session.get(url) as resp:
  8.         text = await resp.text()
  9.         await sleep(2)  # for demo purposes
  10.         print(&quot;Got response from&quot;, url, text.strip().split(&quot;\n&quot;, 1)[0])
  12. async def get_all(urls: list[str], num_concurrent: int) -&gt; None:
  13.     url_iterator = iter(urls)
  14.     keep_going = True
  15.     async with ClientSession() as session:
  16.         while keep_going:
  17.             tasks = []
  18.             for _ in range(num_concurrent):
  19.                 try:
  20.                     url = next(url_iterator)
  21.                 except StopIteration:
  22.                     keep_going = False
  23.                     break
  24.                 new_task = create_task(get_one(session, url))
  25.                 tasks.append(new_task)
  26.             await gather(*tasks)
  28. async def main() -&gt; None:
  29.     urls = [
  30.         &quot;https://github.com&quot;,
  31.         &quot;https://stackoverflow.com&quot;,
  32.         &quot;https://python.org&quot;,
  33.     ]
  34.     await get_all(urls, 2)
  36. run(main())

Output:

  1. Requesting https://github.com
  2. Requesting https://stackoverflow.com
  3. Got response from https://github.com &lt;!DOCTYPE html&gt;
  4. Got response from https://stackoverflow.com &lt;!DOCTYPE html&gt;
  5. Requesting https://python.org
  6. Got response from https://python.org &lt;!doctype html&gt;

You'll notice that the third requests (to python.org) is only sent after both previous requests have returned a response. This setup will essentially perform your total number of requests in batches of num_concurrent.

Python &gt;=3.11

With the newer TaskGroup class, we can make the get_all function a bit more concise:

  1. from asyncio import TaskGroup, run, sleep
  3. from aiohttp import ClientSession
  6. async def get_one(session: ClientSession, url: str) -&gt; None:
  7.     ...  # same as above
  10. async def get_all(urls: list[str], num_concurrent: int) -&gt; None:
  11.     url_iterator = iter(urls)
  12.     keep_going = True
  13.     async with ClientSession() as session:
  14.         while keep_going:
  15.             with TaskGroup() as tg:
  16.                 for _ in range(num_concurrent):
  17.                     try:
  18.                         url = next(url_iterator)
  19.                     except StopIteration:
  20.                         keep_going = False
  21.                         break
  22.                     tg.create_task(get_one(session, url))
  24. ...

Option B: With just asyncio in a Queue

The asyncio.Queue allows us to set a maximum size for it. That makes it possible to limit the maximum number of concurrently executing tasks, but we will need to use the consumer-producer-pattern:

  1. from asyncio import Queue, create_task, gather, run, sleep
  3. from aiohttp import ClientSession
  5. async def get_one(session: ClientSession, url: str) -&gt; None:
  6.     ...  # same as above
  8. STOP_SENTINEL = object()
  10. async def consumer(session: ClientSession, q: Queue[str]) -&gt; None:
  11.     url = await q.get()
  12.     while url is not STOP_SENTINEL:
  13.         await get_one(session, url)
  14.         q.task_done()
  15.         url = await q.get()
  16.     q.task_done()
  18. async def main() -&gt; None:
  19.     urls = [
  20.         &quot;https://github.com&quot;,
  21.         &quot;https://stackoverflow.com&quot;,
  22.         &quot;https://python.org&quot;,
  23.     ]
  24.     num_concurrent = 2
  25.     q = Queue(maxsize=num_concurrent)
  26.     async with ClientSession() as session:
  27.         consumers = [
  28.             create_task(consumer(session, q))
  29.             for _ in range(num_concurrent)
  30.         ]
  31.         for url in urls:
  32.             await q.put(url)
  33.         for _ in range(num_concurrent):
  34.             await q.put(STOP_SENTINEL)
  35.         await gather(*consumers)
  37. run(main())

Output:

  1. Requesting https://github.com
  2. Requesting https://stackoverflow.com
  3. Got response from https://github.com &lt;!DOCTYPE html&gt;
  4. Requesting https://python.org
  5. Got response from https://stackoverflow.com &lt;!DOCTYPE html&gt;
  6. Got response from https://python.org &lt;!doctype html&gt;

As you can see now, that third request can be sent as soon as either of the previous two returns with a response.

That may be more efficient, even though the setup is a bit more cumbersome.

Option C: With an extra package

I used to run into similar issues with setting a fixed number of asyncio tasks to work on a large number of actual tasks. To make this easier I wrote the asyncio-taskpool package. With it I can do something like this:

  1. from asyncio import run, sleep
  3. from aiohttp import ClientSession
  4. from asyncio_taskpool import TaskPool
  6. async def get_one(session: ClientSession, url: str) -&gt; None:
  7.     ...  # same as above
  9. async def get_all(urls: list[str], num_concurrent: int) -&gt; None:
  10.     pool = TaskPool()
  11.     async with ClientSession() as session:
  12.         pool.starmap(
  13.             get_one,
  14.             ((session, url) for url in urls),
  15.             num_concurrent=num_concurrent,
  16.         )
  17.         await pool.gather_and_close()
  19. async def main() -&gt; None:
  20.     urls = [
  21.         &quot;https://github.com&quot;,
  22.         &quot;https://stackoverflow.com&quot;,
  23.         &quot;https://python.org&quot;,
  24.     ]
  25.     await get_all(urls, 2)
  27. run(main())

Output: (same as with the Queue approach)

  1. Requesting https://github.com
  2. Requesting https://stackoverflow.com
  3. Got response from https://github.com &lt;!DOCTYPE html&gt;
  4. Requesting https://python.org
  5. Got response from https://stackoverflow.com &lt;!DOCTYPE html&gt;
  6. Got response from https://python.org &lt;!doctype html&gt;

You'll notice again that the third request will only be made after at least one of the other two returns with a response.

You can try that out with larger numbers of tasks. The number being executed concurrently at any given time will never exceed num_concurrent as passed to map (starmap is just a variant of map).

I tried to emulate the standard multiprocessing.Pool interface to an extent and find this more convenient to use, especially with long-running tasks.

huangapple
  • 本文由 发表于 2023年5月22日 11:05:02
  • 转载请务必保留本文链接：https://go.coder-hub.com/76302813.html
  • asynchronous
  • python
  • python-asyncio
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定