多处理：优化并发 HTTP 异步请求的 CPU 使用率

Posted 2023-03-30

【中文标题】多处理：优化并发 HTTP 异步请求的 CPU 使用率

【英文标题】：Multiprocessing: optimize CPU usage for concurrent HTTP async requests

【发布时间】：2021-03-01 11:01:54

【问题描述】：

我需要下载网站/URL 列表（可能会随时间变化），我目前使用 multiprocessing.Manager().Queue() 提交和更新所述列表。 我必须每秒检查每个 URL/任务：因此每个任务基本上永远不会结束（直到满足特定条件，例如用户中断）。 我认为multiprocessing.Process() 结合asyncio 和一个好的async HTTP client 可以解决问题。不幸的是，在提交 50 个或更多 URL 后，我的 CPU 使用率仍然很高。当任务没有执行任何请求时 - 运行 mock_request() - 和当它们运行时 - 运行 do_request() -时，您自己会注意到差异。

这里是重现每种情况的示例（按 CTRL+C 可随时优雅地结束）。

import asyncio, os, sys, time, httpx
import multiprocessing
import queue as Queue

class ExitHandler(object):
    def __init__(self, manager, queue, processes):
        self.manager = manager
        self.queue = queue
        self.processes = processes
        
    def set_exit_handler(self):
        if os.name == "nt":
            try:
                import win32api
                win32api.SetConsoleCtrlHandler(self.on_exit, True)
            except ImportError:
                version = ".".join(map(str, sys.version_info[:2]))
                raise Exception("pywin32 not installed for Python " + version)
        else:
            import signal
            signal.signal(signal.SIGINT, self.on_exit)
            #signal.signal(signal.CTRL_C_EVENT, func)
            signal.signal(signal.SIGTERM, self.on_exit)

    def on_exit(self, sig, func=None):
        print('[Main process]: exit triggered, terminating all workers')
        STOP_WAIT_SECS= 5 
        for _ in range(N_WORKERS):
            self.queue.put('END')
        
        try:
            end_time = time.time() + STOP_WAIT_SECS
            # wait up to STOP_WAIT_SECS for all processes to complete
            for proc in self.processes:
                join_secs = max(0.0, min(end_time - time.time(), STOP_WAIT_SECS))
                proc.join(join_secs)

            # clear the procs list and _terminate_ any procs that have not yet exited
            while self.processes and len(self.processes) > 0:
                proc = self.processes.pop()
                if proc.is_alive():
                    proc.terminate()
            
            self.manager.shutdown()

            # finally, kill this thread and any running
            os._exit(0)
        except Exception:
            pass

async def mock_request(url):

    # we won't do any request here, it's just an example of how much less CPU
    # each process consumes when not doing requests

    x = 0
    while True:
        try:
            x += 1
            print('Finished downloading '.format(url))
            await asyncio.sleep(1)
        except asyncio.CancelledError:
            return

async def do_request(url):

    while True:
        try:
            # I use httpx (https://github.com/encode/httpx/) as async client for its simplicity
            # feel free to use your preferred library (e.g. aiohttp)
            async with httpx.AsyncClient() as s:
                await s.get(url)
                print('Finished downloading '.format(url))
                await asyncio.sleep(1)
        except asyncio.CancelledError:
            return

def worker(queue):
    
    try:
        event_loop = asyncio.get_event_loop()
        event_loop.run_until_complete(request_worker(queue))
    except KeyboardInterrupt:
        pass

async def request_worker(queue):
    
    p = multiprocessing.current_process()
    loop = asyncio.get_event_loop()
    
    while True:
        try:
            task = await loop.run_in_executor(None, queue.get)
            
            if task == 'END':
                break
            
            elif task['action'] == 'DOWNLOAD':
                print('Worker : Received new task'.format(p.name))
                f = loop.create_task(do_request(task['url'])) # high CPU usage
                # f = loop.create_task(mock_request(task['url'])) # low (almost none) CPU usage

        except KeyboardInterrupt:
            pass
        except Queue.Empty:
            pass

    print('Task Worker : ending'.format(p.name))

def run_workers(queue, processes):

    print('Starting workers')

    for _ in range(N_WORKERS):
        processes.append(multiprocessing.Process(target=worker, args=(queue,)))

    task = 
        'action': 'DOWNLOAD',
        'url': 'https://google.com'
    
    
    # this is just an example forcing the same URL * 100 times, while in reaility
    # it will be 1 different URL per task
    for _ in range(100):
        queue.put(task)

    for p in processes:
        p.start()

    for p in processes:
        p.join()
    
    return True

if __name__ == "__main__":
    processes = []
    N_WORKERS = 8 # processes to spawn
    manager = multiprocessing.Manager()
    q = manager.Queue() # main queue to send URLs to

    # just a useful clean exit handler (press CTRL+C to terminate)
    exit_handler = ExitHandler(manager, q, processes) 
    exit_handler.set_exit_handler()

    # start the workers
    run_workers(q, processes)

这里只是一个例子，说明在同时执行请求时每个进程消耗多少 CPU：

任何显着降低 CPU 使用率（保持每秒相同数量的请求）的解决方案都被接受，无论它是否使用多处理。 对我来说唯一必须是async 模式。

【问题讨论】：

如果你只是......不使用多处理，但仍然使用异步呢？我想这是一个长镜头，但多处理开销可能是 CPU 使用率的重要组成部分。 （即使不是，拥有一个进程可能会使分析更容易）

@tjollans 任何解决方案都被接受，我自己尝试了几种方法 - 包括单个进程。我仍然无法显着降低 CPU 使用率（保持每秒相同数量的请求）。

我没有为我的案例使用多处理，但我在使用 httpx 产生大量请求 (~500 requests \ sec) 时遇到了问题。我最终使用 aiohttp 作为异步 http 客户端后端，因为它可以处理我的情况而不会超时。

【参考方案1】：

这很明显：

while True:
    try:
        async with httpx.AsyncClient() as s:

这会为每个请求初始化一个新客户端，通过查看实现，它会导入并初始化 SSL 上下文。 这些是 IMO 昂贵的操作，因此在循环中运行它们可能会消耗如此多的 CPU。

相反，考虑将代码重新排序为

async with httpx.AsyncClient() as s:
  while True:
    try:

【讨论】：

好吧，我猜你发现了这个错误......我将不得不重新修改我的逻辑，因为这只是一个简单的例子，但我已经做了一些测试，是的，这似乎是问题。我会在几个小时内确认并接受这个答案。