python 异步映射函数，将函数应用于列表的每个元素

Question

我有一个 python 元素列表，我想对其应用 API 调用。 API涉及到Web请求，我想同步应用。由于它是第三方 API，我也无法真正修改里面的代码。我想要这样的东西：

from API import get_user_handle
arr = [1234, 2345, 6789] # list of user id
new_arr = async_foreach(arr, get_user_handle)
print(new_arr) # e.g. ['Tom', 'Amy', 'Jerry']

通过使用数组映射函数，这在nodejs中很容易实现。如何在 python 中构建类似的东西？

Answer 1

您可以使用线程或进程池。例如

import multiprocessing as mp
from API import get_user_handle

if __name__ == "__main__":
    arr = [1234, 2345, 6789] # list of user id
    # guess some reasonable subprocess pool size. This one creates
    # 2 processes per cpu assuming there is I/O wait time. Needs
    # tweeking to environment.
    count = min(len(arr), mp.cpu_count() * 2)
    with mp.Pool(count) as pool:
        new_arr = pool.map(get_user_handle, arr)
    print(new_arr) # e.g. ['Tom', 'Amy', 'Jerry']

线程池的启动速度以及数据传入和传出的速度更快。但Python 的全局解释器锁（GIL）一次只允许一个线程运行。如果一个操作主要是 I/O 密集型，那么它是一个不错的选择。

如果与正在运行的函数的 CPU 负载相比，发送到并行程序或从并行程序发出的数据较小，则进程池就有意义。

Answer 2

使用

concurrent.futures

，我建议在这种情况下使用线程，如果你是 IO 绑定的话：

import concurrent.futures

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    new_arr = executor.map(get_user_handle, arr)

Answer 3

如果我理解你想要正确执行的操作，则可以使用 Python 中的

map

函数来完成此操作。这是一个例子：

import requests

def check_site_up(url):
    res = requests.get(url)
    return res.status_code

sites_to_check = ["https://google.com", "https://youtube.com", "https://yahoo.com"]

results = list(map(check_site_up, sites_to_check))

print(results)

>>> [200, 200, 200]

map

函数是not异步的，所以如果您尝试异步发出请求，您可能需要将其包装在异步函数中，即

from asyncstdlib.builtins import map as amap
import aiohttp
import asyncio

async def anext(ait):
    return await ait.__anext__()

async def check_site_up(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            page = await resp.text()
            status = resp.status
            return status


async def main():
    sites_to_check = ["https://google.com", "https://youtube.com", "https://yahoo.com"]
    results = amap(check_site_up, sites_to_check)
    status_results = list()
    async for res in results:
        status_results.append(res)
    return status_results

status_codes = asyncio.run(main())

print(status_codes)

>>> [200,200,200]