我对发布的大多数答案都有很多问题——他们要么使用了已弃用的库，这些库已经移植了有限的功能，要么提供了一个在执行请求时具有太多魔力的解决方案，使得错误处理变得困难。如果它们不属于上述类别之一，则它们是第三方库或已弃用。

有些解决方案完全适用于http请求，但解决方案不适用于任何其他类型的请求，这是可笑的。这里不需要高度定制的解决方案。

简单地使用python内置库asyncio就足以执行任何类型的异步请求，并为复杂的和特定于用例的错误处理提供足够的流动性。

import asyncio

loop = asyncio.get_event_loop()

def do_thing(params):
    async def get_rpc_info_and_do_chores(id):
        # do things
        response = perform_grpc_call(id)
        do_chores(response)

    async def get_httpapi_info_and_do_chores(id):
        # do things
        response = requests.get(URL)
        do_chores(response)

    async_tasks = []
    for element in list(params.list_of_things):
       async_tasks.append(loop.create_task(get_chan_info_and_do_chores(id)))
       async_tasks.append(loop.create_task(get_httpapi_info_and_do_chores(ch_id)))

    loop.run_until_complete(asyncio.gather(*async_tasks))

它的工作原理很简单。您正在创建一系列希望异步发生的任务，然后请求一个循环执行这些任务并在完成时退出。不需要维护额外的库，也不缺少所需的功能。

上面的答案都没有帮助我，因为他们假设你有一个预定义的请求列表，而在我的情况下，我需要能够侦听请求和异步响应(类似于它在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服务器中的工作方式

我对发布的大多数答案都有很多问题——他们要么使用了已弃用的库，这些库已经移植了有限的功能，要么提供了一个在执行请求时具有太多魔力的解决方案，使得错误处理变得困难。如果它们不属于上述类别之一，则它们是第三方库或已弃用。

有些解决方案完全适用于http请求，但解决方案不适用于任何其他类型的请求，这是可笑的。这里不需要高度定制的解决方案。

简单地使用python内置库asyncio就足以执行任何类型的异步请求，并为复杂的和特定于用例的错误处理提供足够的流动性。

import asyncio

loop = asyncio.get_event_loop()

def do_thing(params):
    async def get_rpc_info_and_do_chores(id):
        # do things
        response = perform_grpc_call(id)
        do_chores(response)

    async def get_httpapi_info_and_do_chores(id):
        # do things
        response = requests.get(URL)
        do_chores(response)

    async_tasks = []
    for element in list(params.list_of_things):
       async_tasks.append(loop.create_task(get_chan_info_and_do_chores(id)))
       async_tasks.append(loop.create_task(get_httpapi_info_and_do_chores(ch_id)))

    loop.run_until_complete(asyncio.gather(*async_tasks))

它的工作原理很简单。您正在创建一系列希望异步发生的任务，然后请求一个循环执行这些任务并在完成时退出。不需要维护额外的库，也不缺少所需的功能。

我已经使用python请求异步调用github的gist API有一段时间了。

举个例子，请看下面的代码:

https://github.com/davidthewatson/flasgist/blob/master/views.py#L60-72

这种风格的python可能不是最清晰的例子，但我可以向您保证代码是有效的。如果这让你感到困惑，请告诉我，我会记录下来。

你可以使用httpx。

import httpx

async def get_async(url):
    async with httpx.AsyncClient() as client:
        return await client.get(url)

urls = ["http://google.com", "http://wikipedia.org"]

# Note that you need an async context to use `await`.
await asyncio.gather(*map(get_async, urls))

如果你想要一个函数式语法，gamla库将其包装到get_async中。

然后你就可以

await gamla.map(gamla.get_async(10))(["http://google.com", "http://wikipedia.org"])

10是超时时间，单位是秒。

(声明:我是作者)

也许请求-期货是另一种选择。

from requests_futures.sessions import FuturesSession

session = FuturesSession()
# first request is started in background
future_one = session.get('http://httpbin.org/get')
# second requests is started immediately
future_two = session.get('http://httpbin.org/get?foo=bar')
# wait for the first request to complete, if it hasn't already
response_one = future_one.result()
print('response one status: {0}'.format(response_one.status_code))
print(response_one.content)
# wait for the second request to complete, if it hasn't already
response_two = future_two.result()
print('response two status: {0}'.format(response_two.status_code))
print(response_two.content)

办公文档中也有建议。如果你不想卷入gevent，这是一个不错的选择。