如何在pythonhttp中设置超时

2024-10-04 05:33:53 发布

您现在位置：Python中文网/ 问答频道 /正文
9078 3

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

我已经试过以下问题中的所有片段：How to implement Timeout in 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()

作为测试，我运行的shell命令需要20秒才能执行，所以我需要在这之前停止服务器。在


Tags： runimportself服务器sendgetserveros
1条回答
网友
1楼 · 发布于 2024-10-04 05:33:53

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

如果可以，可以考虑在服务器前面放置一个代理服务器（比如nginx），让它为您处理超时，或者使用一个更健壮的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()

相关问题 更多 >

    编程相关推荐