我正在收集网站列表上的统计数据，为了简单起见，我正在使用请求。这是我的代码:
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())) )
现在，我想要请求。10秒后进入超时，这样循环就不会卡住。
这个问题以前也很有趣，但没有一个答案是干净的。
我听说可能不使用请求是一个好主意，但我应该如何得到请求提供的好东西(元组中的那些)。
当前回答
我相信你可以使用多处理，而不依赖于第三方包:
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是获取完整响应的超时。
其他回答
我相信你可以使用多处理，而不依赖于第三方包:
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是获取完整响应的超时。
有一个叫做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))
它使用这里的一些答案所建议的信号方法。或者，你可以告诉它使用多处理而不是信号(例如，如果你在多线程环境中)。
我使用请求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):尽管在引发此异常时没有传递任何消息。
尽管问题是关于请求的，但我发现使用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
设置超时参数:
r = requests.get(w, verify=False, timeout=10) # 10 seconds
2.25.1版的更改
如果读取之间的连接或延迟超过10秒，上面的代码将导致对requests.get()的调用超时。见:https://requests.readthedocs.io/en/stable/user/advanced/超时
