设置超时参数:

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

2.25.1版的更改

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

如果遇到这种情况，创建一个看门狗线程，在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)

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

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

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

这里有一些注意事项:

它不是线程安全的，信号总是传递到主线程，所以你不能把它放在任何其他线程中。 在调度信号和执行实际代码之后会有一个轻微的延迟。这意味着示例即使只休眠了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')

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

资料来源及推荐阅读:

关于信号的文档 这是@David Narayan对暂停的回答。他以装饰者的身份组织了上面的代码。

更新:https://requests.readthedocs.io/en/master/user/advanced/超时

在新版本的请求:

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

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

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

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 This code should cause us to set the read timeout as equal to the connect timeout, which is the timeout value you pass on your Session.get() call. (Note that I haven't actually tested this code, so it may need some quick debugging, I just wrote it straight into the GitHub window.)

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) 超时值将应用于连接和读取 超时。如果要设置值，请指定一个元组 另外: R = requests.get('https://github.com'， timeout=(3.05, 27))

Kevinburke已请求将其合并到主要请求项目中，但尚未被接受。

我相信你可以使用多处理，而不依赖于第三方包:

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。我就是不喜欢超时。

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(秒)

由于请求>= 2.4.0，您可以使用timeout参数，即:

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

注意:

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

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)

请注意geevent .timeout. timeout不会被常规异常处理捕获。 所以要么显式地捕获getevent。timeout。timeout 或者传入一个不同的异常，像这样使用:with gevent。Timeout(5, requests.exceptions.Timeout):尽管在引发此异常时没有传递任何消息。

连接超时是请求等待客户端建立到远程机器的连接(对应于套接字上的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年，仅凭请求仍然绝对无法正确地做到这一点。这是库开发人员有意识的设计决定。

利用超时参数的解决方案根本不能完成它们想要做的事情。事实上，乍一看，它“似乎”起作用纯粹是偶然的:

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

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

下面是一些完全打破超时和基于流的方法的示例。都试试。不管你使用哪种方法，它们都是无限期地挂着的。

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

正确的解决方法

My approach utilizes Python's sys.settrace function. It is dead simple. You do not need to use any external libraries or turn your code upside down. Unlike most other answers, this actually guarantees that the code executes in specified time. Be aware that you still need to specify the timeout parameter, as settrace only concerns Python code. Actual socket reads are external syscalls which are not covered by settrace, but are covered by the timeout parameter. Due to this fact, the exact time limit is not TOTAL_TIMEOUT, but a value which is explained in comments below.

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)

就是这样!

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

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

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在后台运行，需要比所有其他导入更早地导入请求，线程等。