时间戳

首页 > 解决方案 > 如何在 Python HTTPServer 中设置超时

python - 如何在 Python HTTPServer 中设置超时

问题描述

我有一个在 Debian 中运行的 Python (2.7.13) HTTP 服务器，我想停止任何花费超过 10 秒的 GET 请求，但在任何地方都找不到解决方案。

我已经尝试了以下问题中发布的所有片段：如何在 BaseHTTPServer.BaseHTTPRequestHandler Python 中实现超时

#!/usr/bin/env python
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
import os

class handlr(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-type','text-html')
        self.end_headers()
        self.wfile.write(os.popen('sleep 20 & echo "this took 20 seconds"').read())

def run():
    server_address = ('127.0.0.1', 8080)
    httpd = HTTPServer(server_address, handlr)
    httpd.serve_forever()

if __name__ == '__main__':
    run()

作为测试，我正在运行一个需要 20 秒才能执行的 shell 命令，因此我需要在此之前停止服务器。

标签： pythontimeouthttpserverbasehttpserverbasehttprequesthandler

解决方案

将您的操作放在后台线程上，然后等待后台线程完成。没有一种通用的安全方法来中止线程，所以不幸的是，这个实现让函数在后台运行，即使它已经放弃了。

如果可以的话，您可能会考虑nginx在您的服务器前面放置一个代理服务器（例如 say ）并让它为您处理超时，或者可能使用更健壮的 HTTP 服务器实现来允许它作为配置选项。但是下面的答案应该基本上涵盖了它。

#!/usr/bin/env python
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
import os
import threading

class handlr(BaseHTTPRequestHandler):
    def do_GET(self):
        result, succeeded = run_with_timeout(lambda: os.popen('sleep 20 & echo "this took 20 seconds"').read(), timeout=3)
        if succeeded:
            self.send_response(200)
            self.send_header('Content-type','text-html')
            self.end_headers()
            self.wfile.write(os.popen('sleep 20 & echo "this took 20 seconds"').read())
        else:
            self.send_response(500)
            self.send_header('Content-type','text-html')
            self.end_headers()
            self.wfile.write('<html><head></head><body>Sad panda</body></html>')
        self.wfile.close()

def run():
    server_address = ('127.0.0.1', 8080)
    httpd = HTTPServer(server_address, handlr)
    httpd.serve_forever()

def run_with_timeout(fn, timeout):
    lock = threading.Lock()
    result = [None, False]
    def run_callback():
        r = fn()
        with lock:
            result[0] = r
            result[1] = True

    t = threading.Thread(target=run_callback)
    t.daemon = True
    t.start()
    t.join(timeout)
    with lock:
        return tuple(result)

if __name__ == '__main__':
    run()

推荐阅读