设置超时参数:

r = requests.get(w, verify=False, timeout=10) # 10 seconds

2.25.1版的更改

如果读取之间的连接或延迟超过10秒，上面的代码将导致对requests.get()的调用超时。看:https://requests.readthedocs.io/en/stable/user/advanced/#timeouts

如果遇到这种情况，创建一个监管机构线程，在10秒后搞乱请求的内部状态，例如:

• 关闭底层套接字，理想情况下
• 如果请求重试该操作，则触发异常

请注意，根据系统库的不同，您可能无法设置DNS解析的截止日期。

这可能有点过分，但是芹菜分布式任务队列对超时有很好的支持。

特别是，您可以定义一个软时间限制，它只在您的流程中引发一个异常(这样您就可以清理)和/或一个硬时间限制，它在超过时间限制时终止任务。

在封面之下，这使用了与你的“之前”帖子中引用的相同的信号方法，但以一种更可用和更易于管理的方式。如果你监控的网站列表很长，你可能会从它的主要功能中受益——各种各样的方法来管理大量任务的执行。

使用eventlet怎么样?如果你想在10秒后超时请求，即使数据正在接收，下面的代码段将为你工作:

import requests
import eventlet
eventlet.monkey_patch()


with eventlet.Timeout(10):
requests.get("http://ipv4.download.thinkbroadband.com/1GB.zip", verify=False)

要创建超时，可以使用信号。

解决这个案子最好的办法可能是

1. 设置一个异常作为告警信号的处理程序
2. 延迟十秒发出警报信号
3. 调用try-except-finally块中的函数。
4. 如果函数超时，则到达except块。
5. 在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)

这里有一些注意事项:

1. 它不是线程安全的，信号总是传递到主线程，所以你不能把它放在任何其他线程中。
2. 在调度信号和执行实际代码之后会有一个轻微的延迟。这意味着示例即使只休眠了10秒也会超时。

但是，这些都在标准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')

这种上下文管理器方法的一个可能的缺点是，您无法知道代码是否实际超时。

资料来源及推荐阅读:

• 信号的相关文档 . b
• 这是@David Narayan对暂停的回答。他将上面的代码组织为装饰。

更新:https://requests.readthedocs.io/en/master/user/advanced/#timeouts

在requests的新版本中:

如果你为超时指定一个单独的值，像这样:

r = requests.get('https://github.com', timeout=5)

超时值将应用于connect和read超时。如果你想分别设置值，请指定一个元组:

r = requests.get('https://github.com', timeout=(3.05, 27))

如果远程服务器非常慢，您可以告诉Requests永远等待响应，方法是将None作为超时值，然后检索一杯咖啡。

r = requests.get('https://github.com', timeout=None)

我以前的答案(可能已经过时了)(很久以前贴出来的):

还有其他方法可以克服这个问题:

1. 使用TimeoutSauce内部类

来自:https://github.com/kennethreitz/requests/issues/1928#issuecomment-35811896

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

这段代码应该导致我们将读超时设置为等于 连接超时，这是传递给 Session.get()调用。(注意，我实际上并没有测试这段代码，所以 它可能需要一些快速调试，我只是把它直接写进 GitHub窗口。)< / p >

2. 使用kevinburke请求的分支: 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)

超时值将应用于连接和读取 超时。如果要设置值，请指定一个元组 另外:< / p >

r = requests.get('https://github.com', timeout=(3.05, 27))

凯文伯克的要求它被合并到主请求项目，但它还没有被接受。

我相信你可以使用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是获取完整的响应的超时。

我想到了一个更直接的解决方案，虽然很难看，但能解决真正的问题。它是这样的:

resp = requests.get(some_url, stream=True)
resp.raw._fp.fp._sock.settimeout(read_timeout)
# This will load the entire response even though stream is set
content = resp.content

你可以阅读完整的解释在这里

此代码工作socketError 11004和10060......

# -*- encoding:UTF-8 -*-
__author__ = 'ACE'
import requests
from PyQt4.QtCore import *
from PyQt4.QtGui import *




class TimeOutModel(QThread):
Existed = pyqtSignal(bool)
TimeOut = pyqtSignal()


def __init__(self, fun, timeout=500, parent=None):
"""
@param fun: function or lambda
@param timeout: ms
"""
super(TimeOutModel, self).__init__(parent)
self.fun = fun


self.timeer = QTimer(self)
self.timeer.setInterval(timeout)
self.timeer.timeout.connect(self.time_timeout)
self.Existed.connect(self.timeer.stop)
self.timeer.start()


self.setTerminationEnabled(True)


def time_timeout(self):
self.timeer.stop()
self.TimeOut.emit()
self.quit()
self.terminate()


def run(self):
self.fun()




bb = lambda: requests.get("http://ipv4.download.thinkbroadband.com/1GB.zip")


a = QApplication([])


z = TimeOutModel(bb, 500)
print 'timeout'


a.exec_()

尽管问题是关于请求的，但我发现这很容易用pycurl CURLOPT_TIMEOUT或CURLOPT_TIMEOUT_MS来完成。

不需要线程或信号:

import pycurl
import StringIO


url = 'http://www.example.com/example.zip'
timeout_ms = 1000
raw = StringIO.StringIO()
c = pycurl.Curl()
c.setopt(pycurl.TIMEOUT_MS, timeout_ms)  # total timeout in milliseconds
c.setopt(pycurl.WRITEFUNCTION, raw.write)
c.setopt(pycurl.NOSIGNAL, 1)
c.setopt(pycurl.URL, url)
c.setopt(pycurl.HTTPGET, 1)
try:
c.perform()
except pycurl.error:
traceback.print_exc() # error generated on timeout
pass # or just pass if you don't want to print the error

嗯，我尝试了这个页面上的许多解决方案，仍然面临不稳定，随机挂起，连接性能差。

我现在正在使用Curl，我对它的“max time”功能和全局性能非常满意，即使实现如此糟糕:

content=commands.getoutput('curl -m6 -Ss "http://mywebsite.xyz"')

这里，我定义了一个最大6秒的时间参数，包括连接时间和传输时间。

我相信Curl有一个很好的python绑定，如果你更喜欢坚持python语法:)

设置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()]

讨论在这里https://redd.it/80kp1h

如果你使用选项stream=True，你可以这样做:

r = requests.get(
'http://url_to_large_file',
timeout=1,  # relevant only for underlying socket
stream=True)


with open('/tmp/out_file.txt'), 'wb') as f:
start_time = time.time()
for chunk in r.iter_content(chunk_size=1024):
if chunk:  # filter out keep-alive new chunks
f.write(chunk)
if time.time() - start_time > 8:
raise Exception('Request took longer than 8s')

该解决方案不需要信号或多处理。

只是另一个解决方案(从http://docs.python-requests.org/en/master/user/advanced/#streaming-uploads得到)

在上传之前，你可以找出内容大小:

TOO_LONG = 10*1024*1024  # 10 Mb
big_url = "http://ipv4.download.thinkbroadband.com/1GB.zip"
r = requests.get(big_url, stream=True)
print (r.headers['content-length'])
# 1073741824


if int(r.headers['content-length']) < TOO_LONG:
# upload content:
content = r.content

但是要小心，发送方可以在“content-length”响应字段中设置不正确的值。

有一个名为timeout-decorator的包，你可以用它来超时任何python函数。

@timeout_decorator.timeout(5)
def mytest():
print("Start")
for i in range(1,10):
time.sleep(1)
print("{} seconds have passed".format(i))

它使用这里的一些答案所建议的信号方法。或者，你可以告诉它使用多处理而不是信号(例如，如果你在多线程环境中)。

timeout = int(seconds)

由于requests >= 2.4.0，你可以使用timeout参数，即:

requests.get('https://duckduckgo.com/', timeout=10)

注意:

timeout不是整个响应下载的时间限制;相反, 如果服务器没有发出响应，则引发exception 超时秒(更准确地说，如果在 底层套接字超时秒)。如果未指定超时时间

.请求不会超时

Timeout =(连接超时，数据读取超时)或给出单个参数(Timeout =1)

import requests


try:
req = requests.request('GET', 'https://www.google.com',timeout=(1,1))
print(req)
except requests.ReadTimeout:
print("READ TIME OUT")

尝试这个请求与超时&错误处理:

import requests
try:
url = "http://google.com"
r = requests.get(url, timeout=10)
except requests.exceptions.Timeout as e:
print e

我使用请求2.2.1和eventlet不适合我。相反，我可以使用gevent超时代替，因为gevent在我的服务中用于gunicorn。

import gevent
import gevent.monkey
gevent.monkey.patch_all(subprocess=True)
try:
with gevent.Timeout(5):
ret = requests.get(url)
print ret.status_code, ret.content
except gevent.timeout.Timeout as e:
print "timeout: {}".format(e.message)

连接超时是number of seconds请求将等待您的客户端建立到远程机器的连接(对应于套接字上的connect()调用)。将连接超时设置为略大于3的倍数是一个很好的实践，3是默认的TCP数据包重传窗口。

客户端连接到服务器并发送HTTP请求后，读取超时启动。它是客户端等待服务器发送响应的秒数。(具体来说，它是客户端在从服务器发送字节之间等待的秒数。在99.9%的情况下，这是服务器发送第一个字节之前的时间)。

如果您为超时指定了一个值，则该超时值将应用于连接超时和读取超时。像下图:

r = requests.get('https://github.com', timeout=5)

如果你想分别设置connect和read的值，请指定一个元组:

r = requests.get('https://github.com', timeout=(3.05, 27))

如果远程服务器非常慢，您可以告诉Requests永远等待响应，方法是将None作为超时值，然后检索一杯咖啡。

r = requests.get('https://github.com', timeout=None)

https://docs.python-requests.org/en/latest/user/advanced/#timeouts

其他答案大多不正确

尽管有这么多的答案，我相信这个帖子仍然缺乏一个合适的解决方案，而且没有现有的答案可以提供一个合理的方法来做一些简单而明显的事情。

让我们从截至2022年，仍然绝对没有办法单独使用requests正确地做到这一点。这是库开发人员的一个有意识的设计决策。开始

使用timeout参数的解决方案根本不能完成它们想要做的事情。事实是它“似乎”;工作乍一看纯属偶然:

timeout参数与请求的总执行时间完全没有关系。它只是控制底层套接字接收任何数据之前可以通过的最大时间量。以5秒的超时为例，服务器也可以每4秒发送1字节的数据，这完全没问题，但对您的帮助不大。

stream和iter_content的答案稍微好一些，但它们仍然不能涵盖请求中的所有内容。在发送响应头之前，你实际上不会从iter_content接收到任何东西，这属于相同的问题——即使你使用1字节作为iter_content的块大小，读取完整的响应头可能需要完全任意的时间，并且你永远无法真正到达从iter_content读取任何响应体的位置。

下面是一些完全破坏基于timeout和__abc1的方法的例子。都试试。不管你使用哪种方法，它们都是无限期地挂着的。

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

浓缩

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)

就是这样!

最大的问题是，如果无法建立连接，requests包会等待太长时间，并阻塞程序的其余部分。

有几种方法来解决这个问题，但当我寻找类似请求的联机程序时，我找不到任何东西。这就是为什么我构建了一个名为reqto的请求包装器(“请求超时”)，它支持来自requests的所有标准方法的适当超时。

pip install reqto

语法与请求相同

import reqto


response = reqto.get(f'https://pypi.org/pypi/reqto/json',timeout=1)
# Will raise an exception on Timeout
print(response)

此外，还可以设置自定义超时函数

def custom_function(parameter):
print(parameter)




response = reqto.get(f'https://pypi.org/pypi/reqto/json',timeout=5,timeout_function=custom_function,timeout_args="Timeout custom function called")
#Will call timeout_function instead of raising an exception on Timeout
print(response)

重要的注意事项是导入行

import reqto

由于monkey_patch在后台运行，需要比所有其他导入更早地导入请求，线程等。