声明:下面的代码为每个函数创建了不同的线程。

这对于某些情况可能是有用的，因为它使用起来更简单。但要知道，它不是异步的，但使用多线程会给人一种异步的错觉，尽管decorator建议这样做。

可以使用以下装饰器在函数执行完成后给出回调，回调必须处理函数返回的数据。

请注意，在函数被修饰后，它将返回一个Future对象。

import asyncio

## Decorator implementation of async runner !!
def run_async(callback, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    def inner(func):
        def wrapper(*args, **kwargs):
            def __exec():
                out = func(*args, **kwargs)
                callback(out)
                return out

            return loop.run_in_executor(None, __exec)

        return wrapper

    return inner

实现示例:

urls = ["https://google.com", "https://facebook.com", "https://apple.com", "https://netflix.com"]
loaded_urls = []  # OPTIONAL, used for showing realtime, which urls are loaded !!


def _callback(resp):
    print(resp.url)
    print(resp)
    loaded_urls.append((resp.url, resp))  # OPTIONAL, used for showing realtime, which urls are loaded !!


# Must provide a callback function, callback func will be executed after the func completes execution
# Callback function will accept the value returned by the function.
@run_async(_callback)
def get(url):
    return requests.get(url)


for url in urls:
    get(url)

如果你想看到实时加载的url，你可以在最后添加以下代码:

while True:
    print(loaded_urls)
    if len(loaded_urls) == len(urls):
        break

我赞同上述使用HTTPX的建议，但我经常以不同的方式使用它，所以我补充了我的答案。

我个人使用asyncio.run(在Python 3.7中引入)而不是asyncio。收集，也更喜欢aiostream方法，它可以与asyncio和httpx结合使用。

就像我刚刚发布的这个例子一样，这种风格对于异步处理一组url很有帮助，尽管(常见的)错误发生了。我特别喜欢这种风格如何阐明响应处理发生在哪里，以及如何简化错误处理(我发现异步调用倾向于提供更多的错误处理)。

发布一个简单的异步发出一堆请求的例子更容易，但通常您还想处理响应内容(用它计算一些东西，可能引用您请求的URL要处理的原始对象)。

这种方法的核心是:

async with httpx.AsyncClient(timeout=timeout) as session:
    ws = stream.repeat(session)
    xs = stream.zip(ws, stream.iterate(urls))
    ys = stream.starmap(xs, fetch, ordered=False, task_limit=20)
    process = partial(process_thing, things=things, pbar=pbar, verbose=verbose)
    zs = stream.map(ys, process)
    return await zs

地点:

Process_thing是一个异步响应内容处理函数 things是输入列表(URL字符串的URL生成器来自于此)，例如对象/字典列表 Pbar是一个进度条(例如tqdm.tqdm)[可选但有用]

所有这些都放在一个async_fetch_urlset异步函数中，然后通过调用一个名为fetch_things的同步“顶级”函数来运行，该函数运行协程[这是async函数返回的内容]并管理事件循环:

def fetch_things(urls, things, pbar=None, verbose=False):
    return asyncio.run(async_fetch_urlset(urls, things, pbar, verbose))

由于作为输入传递的列表(这里是things)可以就地修改，因此可以有效地获得返回的输出(就像我们从同步函数调用中习惯的那样)

不幸的是，据我所知，请求库不具备执行异步请求的能力。您可以在请求周围包装async/await语法，但这将使底层请求的同步性不会降低。如果您想要真正的异步请求，则必须使用其他提供异步请求的工具。其中一个解决方案是aiohttp (Python 3.5.3+)。根据我在Python 3.7 async/await语法中使用它的经验，它工作得很好。下面我写了执行n个web请求的三个实现

使用Python请求库的纯同步请求(sync_requests_get_all) 同步请求(async_requests_get_all)，使用Python 3.7中包装的Python请求库async/await语法和asyncio 一个真正的异步实现(async_aiohttp_get_all)， Python aiohttp库封装在Python 3.7 async/await语法和asyncio中

"""
Tested in Python 3.5.10
"""

import time
import asyncio
import requests
import aiohttp

from asgiref import sync

def timed(func):
    """
    records approximate durations of function calls
    """
    def wrapper(*args, **kwargs):
        start = time.time()
        print('{name:<30} started'.format(name=func.__name__))
        result = func(*args, **kwargs)
        duration = "{name:<30} finished in {elapsed:.2f} seconds".format(
            name=func.__name__, elapsed=time.time() - start
        )
        print(duration)
        timed.durations.append(duration)
        return result
    return wrapper

timed.durations = []


@timed
def sync_requests_get_all(urls):
    """
    performs synchronous get requests
    """
    # use session to reduce network overhead
    session = requests.Session()
    return [session.get(url).json() for url in urls]


@timed
def async_requests_get_all(urls):
    """
    asynchronous wrapper around synchronous requests
    """
    session = requests.Session()
    # wrap requests.get into an async function
    def get(url):
        return session.get(url).json()
    async_get = sync.sync_to_async(get)

    async def get_all(urls):
        return await asyncio.gather(*[
            async_get(url) for url in urls
        ])
    # call get_all as a sync function to be used in a sync context
    return sync.async_to_sync(get_all)(urls)

@timed
def async_aiohttp_get_all(urls):
    """
    performs asynchronous get requests
    """
    async def get_all(urls):
        async with aiohttp.ClientSession() as session:
            async def fetch(url):
                async with session.get(url) as response:
                    return await response.json()
            return await asyncio.gather(*[
                fetch(url) for url in urls
            ])
    # call get_all as a sync function to be used in a sync context
    return sync.async_to_sync(get_all)(urls)


if __name__ == '__main__':
    # this endpoint takes ~3 seconds to respond,
    # so a purely synchronous implementation should take
    # little more than 30 seconds and a purely asynchronous
    # implementation should take little more than 3 seconds.
    urls = ['https://postman-echo.com/delay/3']*10

    async_aiohttp_get_all(urls)
    async_requests_get_all(urls)
    sync_requests_get_all(urls)
    print('----------------------')
    [print(duration) for duration in timed.durations]

在我的机器上，这是输出:

async_aiohttp_get_all          started
async_aiohttp_get_all          finished in 3.20 seconds
async_requests_get_all         started
async_requests_get_all         finished in 30.61 seconds
sync_requests_get_all          started
sync_requests_get_all          finished in 30.59 seconds
----------------------
async_aiohttp_get_all          finished in 3.20 seconds
async_requests_get_all         finished in 30.61 seconds
sync_requests_get_all          finished in 30.59 seconds

from threading import Thread

threads=list()

for requestURI in requests:
    t = Thread(target=self.openURL, args=(requestURI,))
    t.start()
    threads.append(t)

for thread in threads:
    thread.join()

...

def openURL(self, requestURI):
    o = urllib2.urlopen(requestURI, timeout = 600)
    o...

上面的答案都没有帮助我，因为他们假设你有一个预定义的请求列表，而在我的情况下，我需要能够侦听请求和异步响应(类似于它在nodejs中的工作方式)。

def handle_finished_request(r, **kwargs):
    print(r)


# while True:
def main():
    while True:
        address = listen_to_new_msg()  # based on your server

        # schedule async requests and run 'handle_finished_request' on response
        req = grequests.get(address, timeout=1, hooks=dict(response=handle_finished_request))
        job = grequests.send(req)  # does not block! for more info see https://stackoverflow.com/a/16016635/10577976


main()

handle_finished_request回调函数将在收到响应时被调用。注意:由于某些原因，超时(或无响应)在这里不会触发错误

这个简单的循环可以触发异步请求，类似于它在nodejs服务器中的工作方式

我也尝试过使用python中的异步方法做一些事情，然而我使用twisted进行异步编程的运气要好得多。它的问题较少，并且有良好的文档记录。这里有一个类似于你在twisted中尝试的东西的链接。

http://pythonquirks.blogspot.com/2011/04/twisted-asynchronous-http-request.html