Python aiohttp (with asyncio) 发送请求很慢

Question

情况:我正在尝试向我已下载的特定文件中的所有列出的域发送 HTTP 请求，并获取我被转发到的目标 URL。

问题:好吧，我关注了一个 tutorial我收到的回复比预期的少很多。大约是每秒 100 个响应，但教程中列出了每分钟 100,000 个响应。几秒钟后脚本也变得越来越慢，所以我每 5 秒只收到 1 个响应。

已经尝试过: 首先我认为这个问题是因为我在 Windows 服务器上运行它。在我的计算机上尝试该脚本后，我意识到它只是快了一点点，但并不多。在其他 Linux 服务器上，它与在我的计算机(Unix、macOS)上一样。

代码: https://pastebin.com/WjLegw7K

work_dir = os.path.dirname(__file__)

async def fetch(url, session):
    try:
        async with session.get(url, ssl=False) as response:
            if response.status == 200:
                delay = response.headers.get("DELAY")
                date = response.headers.get("DATE")
                print("{}:{} with delay {}".format(date, response.url, delay))
                return await response.read()
    except Exception:
        pass

async def bound_fetch(sem, url, session):
    # Getter function with semaphore.
    async with sem:
        await fetch(url, session)


async def run():
    os.chdir(work_dir)
    for file in glob.glob("cdx-*"):
        print("Opening: " + file)
        opened_file = file
        tasks = []
        # create instance of Semaphore
        sem = asyncio.Semaphore(40000)
        with open(work_dir + '/' + file) as infile:
            seen = set()
            async with ClientSession() as session:
                for line in infile:
                    regex = re.compile(r'://(.*?)/')
                    domain = regex.search(line).group(1)
                    domain = domain.lower()

                    if domain not in seen:
                        seen.add(domain)

                        task = asyncio.ensure_future(bound_fetch(sem, 'http://' + domain, session))
                        tasks.append(task)

                    del line
                responses = asyncio.gather(*tasks)
                await responses
            infile.close()
            del seen
            del file


loop = asyncio.get_event_loop()

future = asyncio.ensure_future(run())
loop.run_until_complete(future)

我真的不知道如何解决这个问题。特别是因为我是 Python 的新手......但我必须以某种方式让它工作:(

Answer 1

如果不实际调试代码，很难说出哪里出了问题，但一个潜在的问题是文件处理是序列化的。换句话说，在当前文件的所有请求完成之前，代码永远不会处理下一个文件。如果有很多文件并且其中一个文件很慢，这可能是个问题。

要更改此设置，请沿以下行定义 run:

async def run():
    os.chdir(work_dir)
    async with ClientSession() as session:
        sem = asyncio.Semaphore(40000)
        seen = set()
        pending_tasks = set()
        for f in glob.glob("cdx-*"):
            print("Opening: " + f)
            with open(f) as infile:
                lines = list(infile)
            for line in lines:
                domain = re.search(r'://(.*?)/', line).group(1)
                domain = domain.lower()
                if domain in seen:
                    continue
                seen.add(domain)
                task = asyncio.ensure_future(bound_fetch(sem, 'http://' + domain, session))
                pending_tasks.add(task)
                # ensure that each task removes itself from the pending set
                # when done, so that the set doesn't grow without bounds
                task.add_done_callback(pending_tasks.remove)
        # await the remaining tasks
        await asyncio.wait(pending_tasks)

另一件重要的事情:在 fetch() 中屏蔽所有异常是不好的做法，因为没有迹象表明某些事情已经开始出错(由于错误或简单的拼写错误)。这很可能是您的脚本在一段时间后变得“缓慢”的原因 - fetch 引发异常，而您永远看不到它们。代替 pass，使用类似 print(f'failed to get {url}: {e}') 的方式，其中 e 是您要访问的对象从 except Exception as e 获取。

一些补充说明:

• Python 中几乎不需要del 局部变量；垃圾收集器会自动执行此操作。
• 您不需要close() 使用with 语句打开的文件。 with 专为您自动关闭而设计。
• 代码将域添加到seen 集，但也处理了一个已经看到的域。此版本跳过它已经为其生成任务的域。
• 您可以创建单个 ClientSession 并在整个运行过程中使用它。