我尝试了Python的requests库文档中提供的示例。
使用async.map(rs)
,我可以获取响应代码,但我想要获取每个请求页面的内容。例如,下面的方式不起作用:
out = async.map(rs)
print out[0].content
我尝试了Python的requests库文档中提供的示例。
使用async.map(rs)
,我可以获取响应代码,但我想要获取每个请求页面的内容。例如,下面的方式不起作用:
out = async.map(rs)
print out[0].content
以下的答案不适用于请求库v0.13.0及以上版本。该异步功能已经被移至grequests,因此该问题被提出后。但是,您可以将下面的requests
替换为grequests
,然后就可以正常工作了。
我将保留此答案,以反映最初的问题,即如何使用请求库小于v0.13.0。
要使用async.map
异步地执行多个任务,您需要:
async.map
以处理所有请求/操作的列表示例:
from requests import async
# If using requests > v0.13.0, use
# from grequests import async
urls = [
'http://python-requests.org',
'http://httpbin.org',
'http://python-guide.org',
'http://kennethreitz.com'
]
# A simple task to do to each response object
def do_something(response):
print response.url
# A list to hold our things to do via async
async_list = []
for u in urls:
# The "hooks = {..." part is where you define what you want to do
#
# Note the lack of parentheses following do_something, this is
# because the response will be used as the first argument automatically
action_item = async.get(u, hooks = {'response' : do_something})
# Add the task to our list of things to do via async
async_list.append(action_item)
# Do our list of things to do via async
async.map(async_list)
from grequests import async
无效,而这个函数定义 def do_something(response, **kwargs):
对我有用。我是从 https://dev59.com/rHDXa4cB1Zd3GeqP7w3A 找到的。 - Allan Ruinfrom requests import async
و›؟وچ¢ن¸؛import grequests as async
ه¯¹وˆ‘وœ‰و•ˆم€‚ - Martin Thomagrequests
现在建议使用requests-threads
或requests-futures
。 - OrangeDogasync
现在是一个独立模块:grequests
。
请参见这里:https://github.com/spyoungtech/grequests
还有这里:Ideal method for sending multiple HTTP requests over Python?
$ pip install grequests
构建一个堆栈:
import grequests
urls = [
'http://www.heroku.com',
'http://tablib.org',
'http://httpbin.org',
'http://python-requests.org',
'http://kennethreitz.com'
]
rs = (grequests.get(u) for u in urls)
grequests.map(rs)
结果看起来像
[<Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>]
results = grequests.map(rs)
,那么这行代码之后的代码就会被阻塞,我能看到异步效果吗? - Allan Ruin我测试了两个库:requests-futures 和 grequests。grequests更快,但会带来猴子补丁和额外的依赖问题;requests-futures比grequests慢好几倍。于是我决定自己写代码,只需将requests包装到ThreadPoolExecutor即可,速度几乎与grequests一样,而且没有外部依赖。
import requests
import concurrent.futures
def get_urls():
return ["url1","url2"]
def load_url(url, timeout):
return requests.get(url, timeout = timeout)
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
future_to_url = {executor.submit(load_url, url, 10): url for url in get_urls()}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
except Exception as exc:
resp_err = resp_err + 1
else:
resp_ok = resp_ok + 1
很不幸,据我所知,requests库并不支持执行异步请求。您可以在requests
周围包装async/await
语法,但这仍然会使底层请求变得同步。如果您想要真正的异步请求,必须使用其他提供此功能的工具。其中之一是aiohttp
(Python 3.5.3+)。我使用Python 3.7的async/await
语法时,它表现良好。以下是使用以下三种方法执行 n 次网络请求:
sync_requests_get_all
),使用Python requests
库async/await
语法和asyncio
包装的同步请求(async_requests_get_all
),使用Python requests
库async/await
语法和asyncio
包装的真正异步实现(async_aiohttp_get_all
),使用Python aiohttp
库"""
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
async_aiohttp_get_all()
是一个不错的解决方案。我也想到了类似的东西,但是在它外面多了一个 async def fetch_all(urls): return await asyncio.gather(*[fetch(url) for url in urls])
,这导致我的解决方案为每个URL创建单独的 aiohttp.ClientSession()
实例,而通过嵌入本地函数,您可以重用相同的会话……在我看来更具有 Python 风格。您能否提醒我使用 sync.async_to_sync()
与存在 get_all()
相比,使用 asyncio.run()
without get_all()
的好处是什么? - wescpyaiohttp
版本是不是只有我一个人觉得它使用了asgiref.sync.async_to_sync
来运行?有没有一种方法可以在不包含额外模块的情况下完成这个操作? - CpILLfrom 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,这是一个好的选择。
ThreadPoolExecutor(max_workers=10)
- Dreampuf您可以使用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
是超时时间,单位为秒。
(免责声明:我是作者)
await asyncio.gather(*map(get_async, urls)) ^ SyntaxError: invalid syntax
请指导。 - AJ.await
。 - Uri我对大多数发布的答案有很多问题 - 它们要么使用已经被移植过来并带有有限功能的废弃库,要么提供了在请求执行过程中具有太多魔法的解决方案,使得错误处理变得困难。如果它们不属于以上任一类别,则是第三方库或已过时的。
其中一些解决方案仅适用于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))
它的工作原理很简单。您正在创建一系列希望异步发生的任务,然后要求循环执行这些任务,并在完成后退出。没有额外的库需要维护,也没有必要缺乏功能。
await response = requests.get(URL)
的代码。是这样吗? - Coder Nr 23requests
进行包装几乎比同步调用URL列表要慢(在某些情况下甚至更慢)。例如,使用上述策略请求一个响应时间为3秒的端点10次大约需要30秒。如果您想要真正的异步性能,您需要使用像aiohttp
这样的东西。 - DragonBobZsynchronous calls
中的requests.get
,你仍然会看到加速。但问题是如何使用Python requests
库执行异步请求。这个答案没有解决这个问题,所以我的批评依旧。 - DragonBobZ我知道这篇文章已经关闭了一段时间,但是我认为推广另一个基于requests库构建的异步解决方案可能会很有用。
list_of_requests = ['http://moop.com', 'http://doop.com', ...]
from simple_requests import Requests
for response in Requests().swarm(list_of_requests):
print response.content
免责声明:以下代码为每个函数创建不同的线程。
这对于某些情况可能很有用,因为它更简单易用。但请注意,它不是异步的,而是使用多个线程给出异步的假象,尽管装饰器表明了这一点。
您可以使用以下装饰器,在函数执行完成后提供回调,回调必须处理函数返回的数据的处理。
请注意,函数装饰后将返回一个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)
while True:
print(loaded_urls)
if len(loaded_urls) == len(urls):
break