使用多进程和多线程提高asyncio"发送100,000请求"的速度

Question

首先,我想尽快使用1个连接发送多个请求.下面的代码运行良好而快速,但我希望它超越异步.回到我的问题,是否可以使用多线程或多处理并行运行它.我听说你可以使用ThreadPoolExecutor或ProcessPoolExecutor.

import random
import asyncio
from aiohttp import ClientSession
import time
from concurrent.futures import ProcessPoolExecutor

async def fetch(sem,url, session):
    async with sem:
        async with session.get(url) as response:
            return await response.read()
async def run(r):
    url = "http://www.example.com/"
    tasks = []
    sem = asyncio.Semaphore(1000)
    async with ClientSession() as session:
        for i in range(r):
            task = asyncio.ensure_future(fetch(sem, url.format(i), session)) #return a task
            tasks.append(task)
        responses = asyncio.gather(*tasks)
        await responses
if __name__ == "__main__":
    number = 10000
    loop = asyncio.get_event_loop()
    start = time.time()
    loop.run_until_complete(run(number))
    end = time.time() - start
    print (end)

从测试开始,它设法在49秒内发送了大约10k的请求.我需要它更快,有什么建议吗？(线程,过程)

Answer 1

ProcessPoolExecutor 是一种进行真正的多处理的方法。对于您的用例，基本上就像您同时启动程序的多个副本一样。如果您的计算机具有所需的带宽和 CPU，则应该能够通过使用 ProcessPoolExecutor(max_workers=4) 将性能提高 4 倍

但是，您将需要在每个子流程中都有一个异步事件循环，因此您可以执行以下操作：

def main(n):
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run(n))


with concurrent.futures.ProcessPoolExecutor(max_workers=4) as exc:
    exc.submit(main, 2500)
    exc.submit(main, 2500)
    exc.submit(main, 2500)
    exc.submit(main, 2500)

作为函数的旁注run：您也不需要使用ensure_future或任务，函数的结果async def是一个协程，您可以直接等待或传递给它asyncio.gather

async def run(r):
    url = "http://www.example.com/"
    sem = asyncio.Semaphore(1000)
    async with ClientSession() as session:
        coros = [fetch(sem, url.format(i), session) for i in range(r)]
        await asyncio.gather(*coros)