我正在搜集一些网站的统计数据，并且出于简便考虑，我正在使用 requests 库。以下是我的代码：
data=[]
websites=['http://google.com', 'http://bbc.co.uk']
for w in websites:
r= requests.get(w, verify=False)
data.append( (r.url, len(r.content), r.elapsed.total_seconds(), str([(l.status_code, l.url) for l in r.history]), str(r.headers.items()), str(r.cookies.items())) )
requests.get 在 10 秒钟后超时，以便循环不会卡住。try:
r = requests.get("MYURL.com", timeout=10) # 10 seconds
except requests.exceptions.Timeout:
print("Timed out")
requests.get()在连接或读取之间的延迟超过十秒时超时。
timeout参数接受等待的秒数，可以是浮点数，也可以是一个(连接超时, 读取超时)元组。使用eventlet如何呢？如果您想在收到数据的情况下在10秒后超时请求，那么这段代码会对您有帮助：
import requests
import eventlet
eventlet.monkey_patch()
with eventlet.Timeout(10):
requests.get("http://ipv4.download.thinkbroadband.com/1GB.zip", verify=False)
eventlet.monkey_patch()吗？ - Userrequests.get('https://github.com', timeout=5) **。 - Pedro Lobito更新：https://requests.readthedocs.io/en/master/user/advanced/#timeouts
在新版本的requests中：
如果您指定一个超时的单一值，如下所示：
r = requests.get('https://github.com', timeout=5)
超时值将应用于连接和读取超时。如果您想分别设置这些值，请指定一个元组：
r = requests.get('https://github.com', timeout=(3.05, 27))
r = requests.get('https://github.com', timeout=None)
我的旧回答（可能已过时），如下：
有其他方法可以解决这个问题：
1. 使用TimeoutSauce内部类
来源：https://github.com/kennethreitz/requests/issues/1928#issuecomment-35811896
这段代码应该会把读取超时设为连接超时的值，这个超时的值是你在Session.get()调用中传递的参数。（注意，我实际上没有测试过这段代码，所以可能需要一些快速调试，我只是直接在GitHub窗口中编写了它。）import requests from requests.adapters import TimeoutSauce class MyTimeout(TimeoutSauce): def __init__(self, *args, **kwargs): connect = kwargs.get('connect', 5) read = kwargs.get('read', connect) super(MyTimeout, self).__init__(connect=connect, read=read) requests.adapters.TimeoutSauce = MyTimeout
2. 使用 kevinburke 的 requests 分支：https://github.com/kevinburke/requests/tree/connect-timeout 从文档中可以看到：https://github.com/kevinburke/requests/blob/connect-timeout/docs/user/advanced.rst如果你为超时指定单个值，像这样：
r = requests.get('https://github.com', timeout=5)超时值将适用于连接超时和读取超时。如果您想分别设置这些值，请指定一个元组：
r = requests.get('https://github.com', timeout=(3.05, 27))kevinburke已要求将其合并到主要的requests项目中, 但尚未被接受。
timeout = int(seconds)自requests >= 2.4.0版本开始，您可以使用timeout参数，例如：
requests.get('https://duckduckgo.com/', timeout=10)
注意:
timeout不是整个响应下载的时间限制; 相反，如果服务器在超时秒内未发出响应(更准确地说，如果底层套接字在超时秒内没有接收到任何字节)，则会引发异常。如果没有显式指定超时，则请求不会超时。
import requests, sys, time
TOTAL_TIMEOUT = 10
def trace_function(frame, event, arg):
if time.time() - start > TOTAL_TIMEOUT:
raise Exception('Timed out!')
return trace_function
start = time.time()
sys.settrace(trace_function)
try:
res = requests.get('http://localhost:8080', timeout=(3, 6))
except:
raise
finally:
sys.settrace(None)
尽管有很多答案，但我认为这个主题仍然缺乏一个合适的解决方案，没有现有的答案提供一种简单明显的方法。
首先要说的是，截至2023年，仅使用requests是绝对无法正确执行它的。 这是库开发人员的有意设计决策。
使用timeout参数的解决方案根本无法实现其预期目的。事实上，它在第一眼看起来“似乎”可以工作纯属偶然：
timeout参数与请求的总执行时间毫不相关。它仅仅控制了底层套接字在接收到任何数据之前可以经过的最长时间。例如，设定5秒的超时时间，服务器也可以每4秒发送1字节的数据，这完全没有问题，但并不能帮助你太多。
使用stream和iter_content的答案要好一些，但它们仍然不能涵盖请求中的所有内容。直到响应头被发送后，才会从iter_content实际接收到任何内容，这也属于同样的问题 - 即使你使用1字节作为iter_content的块大小，读取完整的响应头可能需要完全随意的时间，你永远无法真正到达从iter_content读取任何响应正文的点。
以下是一些完全破坏timeout和基于stream的方法的示例。尝试它们所有。无论你使用哪种方法，它们都会无限期地挂起。
server.py
import socket
import time
server = socket.socket()
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
server.bind(('127.0.0.1', 8080))
server.listen()
while True:
try:
sock, addr = server.accept()
print('Connection from', addr)
sock.send(b'HTTP/1.1 200 OK\r\n')
# Send some garbage headers very slowly but steadily.
# Never actually complete the response.
while True:
sock.send(b'a')
time.sleep(1)
except:
pass
demo1.py
import requests
requests.get('http://localhost:8080')
demo2.py
import requests
requests.get('http://localhost:8080', timeout=5)
demo3.py
import requests
requests.get('http://localhost:8080', timeout=(5, 5))
demo4.py
import requests
with requests.get('http://localhost:8080', timeout=(5, 5), stream=True) as res:
for chunk in res.iter_content(1):
break
我的方法利用Python的sys.settrace函数。它非常简单。你不需要使用任何外部库或改变你的代码结构。与大多数其他答案不同的是，这实际上保证了代码在指定时间内执行。请注意，你仍然需要指定timeout参数，因为settrace只涉及Python代码。实际的套接字读取是外部系统调用，不受settrace覆盖，但受timeout参数覆盖。由于这个事实，确切的时间限制不是TOTAL_TIMEOUT，而是一个在下面的注释中解释的值。
import requests
import sys
import time
# This function serves as a "hook" that executes for each Python statement
# down the road. There may be some performance penalty, but as downloading
# a webpage is mostly I/O bound, it's not going to be significant.
def trace_function(frame, event, arg):
if time.time() - start > TOTAL_TIMEOUT:
raise Exception('Timed out!') # Use whatever exception you consider appropriate.
return trace_function
# The following code will terminate at most after TOTAL_TIMEOUT + the highest
# value specified in `timeout` parameter of `requests.get`.
# In this case 10 + 6 = 16 seconds.
# For most cases though, it's gonna terminate no later than TOTAL_TIMEOUT.
TOTAL_TIMEOUT = 10
start = time.time()
sys.settrace(trace_function)
try:
res = requests.get('http://localhost:8080', timeout=(3, 6)) # Use whatever timeout values you consider appropriate.
except:
raise
finally:
sys.settrace(None) # Remove the time constraint and continue normally.
# Do something with the response
就是这样！
要创建一个超时，你可以使用信号。
解决这个问题的最好方法可能是：
try-except-finally块中调用函数。下面是一些示例代码:
import signal
from time import sleep
class TimeoutException(Exception):
""" Simple Exception to be called on timeouts. """
pass
def _timeout(signum, frame):
""" Raise an TimeoutException.
This is intended for use as a signal handler.
The signum and frame arguments passed to this are ignored.
"""
# Raise TimeoutException with system default timeout message
raise TimeoutException()
# Set the handler for the SIGALRM signal:
signal.signal(signal.SIGALRM, _timeout)
# Send the SIGALRM signal in 10 seconds:
signal.alarm(10)
try:
# Do our code:
print('This will take 11 seconds...')
sleep(11)
print('done!')
except TimeoutException:
print('It timed out!')
finally:
# Abort the sending of the SIGALRM signal:
signal.alarm(0)
但是，这里有一些注意事项：
但是，所有这些都在标准的Python库中！除了导入sleep函数之外，只需要一个导入。如果您要在许多地方使用超时，可以轻松将TimeoutException、_timeout和singaling放在一个函数中，然后只需调用该函数。或者，您可以将其设置为装饰器，并将其放在函数上，请参见下面链接的答案。
您还可以将其设置为“上下文管理器”，以便您可以在with语句中使用它：
import signal
class Timeout():
""" Timeout for use with the `with` statement. """
class TimeoutException(Exception):
""" Simple Exception to be called on timeouts. """
pass
def _timeout(signum, frame):
""" Raise an TimeoutException.
This is intended for use as a signal handler.
The signum and frame arguments passed to this are ignored.
"""
raise Timeout.TimeoutException()
def __init__(self, timeout=10):
self.timeout = timeout
signal.signal(signal.SIGALRM, Timeout._timeout)
def __enter__(self):
signal.alarm(self.timeout)
def __exit__(self, exc_type, exc_value, traceback):
signal.alarm(0)
return exc_type is Timeout.TimeoutException
# Demonstration:
from time import sleep
print('This is going to take maximum 10 seconds...')
with Timeout(10):
sleep(15)
print('No timeout?')
print('Done')
使用这种上下文管理器方法的一个可能的缺点是您无法确定代码是否实际超时。
来源和推荐阅读：
尝试使用超时和错误处理进行此请求：
import requests
try:
url = "http://google.com"
r = requests.get(url, timeout=10)
except requests.exceptions.Timeout as e:
print e
requests.exceptions.ConnectionError而不是Timeout，因此您可能希望捕获一个更通用的RequestException。 - egor83一旦您的客户端连接到服务器并发送了HTTP请求，读取超时时间开始生效。它是客户端等待服务器发送响应的秒数。（具体来说，它是客户端在接收到来自服务器的字节之间等待的时间，几乎所有情况下，这是服务器发送第一个字节前的时间）。
如果您为超时时间指定单个值，则该超时值将同时应用于连接和读取超时。
r = requests.get('https://github.com', timeout=5)
如果您想为连接和读取分别设置值，请指定一个元组：
r = requests.get('https://github.com', timeout=(3.05, 27))
r = requests.get('https://github.com', timeout=None)
https://docs.python-requests.org/zh_CN/latest/user/advanced/#timeouts
设置 stream=True 并使用 r.iter_content(1024)。是的，eventlet.Timeout 对我来说似乎不起作用。
try:
start = time()
timeout = 5
with get(config['source']['online'], stream=True, timeout=timeout) as r:
r.raise_for_status()
content = bytes()
content_gen = r.iter_content(1024)
while True:
if time()-start > timeout:
raise TimeoutError('Time out! ({} seconds)'.format(timeout))
try:
content += next(content_gen)
except StopIteration:
break
data = content.decode().split('\n')
if len(data) in [0, 1]:
raise ValueError('Bad requests data')
except (exceptions.RequestException, ValueError, IndexError, KeyboardInterrupt,
TimeoutError) as e:
print(e)
with open(config['source']['local']) as f:
data = [line.strip() for line in f.readlines()]
The discussion is here https://redd.it/80kp1h
我相信你可以使用multiprocessing而不依赖第三方包：
import multiprocessing
import requests
def call_with_timeout(func, args, kwargs, timeout):
manager = multiprocessing.Manager()
return_dict = manager.dict()
# define a wrapper of `return_dict` to store the result.
def function(return_dict):
return_dict['value'] = func(*args, **kwargs)
p = multiprocessing.Process(target=function, args=(return_dict,))
p.start()
# Force a max. `timeout` or wait for the process to finish
p.join(timeout)
# If thread is still active, it didn't finish: raise TimeoutError
if p.is_alive():
p.terminate()
p.join()
raise TimeoutError
else:
return return_dict['value']
call_with_timeout(requests.get, args=(url,), kwargs={'timeout': 10}, timeout=60)
kwargs 参数中传递的超时时间是从服务器获取 任何 响应的超时时间，而 timeout 参数则是获取完整响应的超时时间。