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

import requests
import eventlet
eventlet.monkey_patch()

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

我使用请求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):尽管在引发此异常时没有传递任何消息。

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

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

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

只是另一个解决方案(从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”响应字段中设置不正确的值。

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

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

设置一个异常作为告警信号的处理程序 延迟十秒发出警报信号 在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对暂停的回答。他以装饰者的身份组织了上面的代码。

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

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是获取完整响应的超时。