Python中HTTPServer多次启停失败问题求助

问题概述 #

实现了一个用于切换状态变量的HTTPServer，单次启停正常，偶尔能完成两次启停，但多次启停后会因超时抛出ConnectionError，报错显示无法连接到服务器，读取超时。

核心错误原因 #

1. 未彻底释放Socket资源：原代码中调用super().shutdown()仅停止了服务器的请求循环，但未关闭底层的监听Socket，导致端口资源未完全释放，后续启动时无法快速重新绑定。
2. 服务器启动后立即请求的竞态条件：serveInNewThread中线程启动后立即发起requests.get，此时服务器可能尚未完成初始化，导致请求超时。
3. 线程资源未彻底清理：原关闭逻辑中仅等待线程结束，但未确保Socket等资源被GC回收。

修复后的完整代码 #

实现代码 #

from typing import Tuple, Type, Optional
import requests
import logging
import threading
import time
from enum import Enum
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import BaseServer

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)


class Status(Enum):
    START_STATE = 0
    END_STATE = 1


class StatusTracker:
    def __init__(self):
        self._status = Status.START_STATE

    @property
    def status(self):
        return self._status

    @status.setter
    def status(self, newStatus):
        self._status = newStatus


class HttpHandler(BaseHTTPRequestHandler):
    """Simple request handler -- toggles the statusTracker.status variable."""

    def __init__(self, request, client_address: Tuple[str, int], server: BaseServer, status_tracker):
        self.status_tracker: Optional[StatusTracker] = status_tracker
        super().__init__(request=request, client_address=client_address, server=server)

    def log_message(self, format, *args):
        # 禁用默认日志，统一使用自定义logger
        logger.debug(f"Client {self.client_address[0]}: {format % args}")

    def do_GET(self):
        logger.debug(f"Handling GET request: {self.path}")

        self.send_response(200)

        if self.path.endswith('/'):
            # Mini landing page for browser access
            self.send_header("Content-type", "text/html")
            self.end_headers()
            landing = """<html>
            <p>Request: {path}</p>
            <body>
            <p>Listening to state change requests.</p>
            </body>
            </html>""".format(path=self.path)
            self.wfile.write(landing.strip().encode("utf-8"))
            return

        # Respond with plain text for API requests
        self.send_header("Content-type", "text/plain")
        self.end_headers()
        logger.info(f"HTTPServer received request: \"{self.path}\"")

        if self.path == '/favicon.ico':
            return

        if self.path.endswith("/test_connection"):
            response_content = "HttpServer is running and responsive"
            self.wfile.write(response_content.encode())
            return
        elif self.path.endswith("/start_state"):
            self.status_tracker.status = Status.START_STATE
        elif self.path.endswith("/end_state"):
            self.status_tracker.status = Status.END_STATE
        else:
            logger.warning(f"Unhandled request: \"{self.path}\"")


class MyHttpServer(HTTPServer):
    """Custom HTTPServer with thread management and proper shutdown logic"""

    def __init__(self, address: Tuple[str, int], handler: Type[HttpHandler],
                 status_tracker: Optional[StatusTracker] = None):
        self.allow_reuse_address = True
        self.allow_reuse_port = True
        self.status_tracker = status_tracker
        super().__init__(address, handler)

        self.server_thread: threading.Thread
        self.server_started_event: threading.Event = threading.Event()
        self.server_finished_event: threading.Event = threading.Event()

    def finish_request(self, request, client_address) -> None:
        assert self.status_tracker is not None
        self.RequestHandlerClass(request, client_address, self, self.status_tracker)

    def serve_forever(self, poll_interval: float = 0.5) -> None:
        try:
            super().serve_forever(poll_interval)
        except Exception as e:
            logger.error(f"Server error: {str(e)}")
            self.shutdown_from_new_thread()

    def serve_in_new_thread(self) -> threading.Thread:
        logger.info(f"Starting server on {self.server_address}")
        self.server_thread = threading.Thread(target=self.serve_forever, name="HttpServerThread", daemon=True)
        self.server_thread.start()

        # 循环等待服务器响应，避免启动立即请求的竞态
        max_retries = 10
        retry_count = 0
        while retry_count < max_retries:
            try:
                response = requests.get(f"http://{self.server_address[0]}:{self.server_address[1]}/test_connection",
                                        timeout=1)
                response.raise_for_status()
                logger.info(response.text)
                self.server_started_event.set()
                self.server_finished_event.clear()
                return self.server_thread
            except requests.exceptions.RequestException:
                retry_count += 1
                time.sleep(0.5)

        raise ConnectionError("Could not connect to HttpServer after multiple attempts")

    def shutdown(self) -> None:
        logger.info("Initiating server shutdown")
        if self._BaseServer__is_shut_down.is_set():
            logger.info("Server is already shut down")
            return

        super().shutdown()
        # 关闭底层Socket，彻底释放端口资源
        self.server_close()
        logger.info("Server socket closed")

        # 等待线程结束
        if self.server_thread.is_alive():
            self.server_thread.join(timeout=5)
            if self.server_thread.is_alive():
                logger.warning("Server thread did not terminate within timeout")

        self.server_finished_event.set()
        self.server_started_event.clear()
        logger.info("Server shutdown complete")

    def shutdown_from_new_thread(self):
        thread = threading.Thread(target=self.shutdown, name="ShutdownHttpServerThread", daemon=True)
        thread.start()
        self.server_finished_event.wait()


class UserOfMyHttpServer(StatusTracker):
    """Implements StatusTracker and manages the HTTP server lifecycle"""

    def __init__(self, port: int):
        super().__init__()
        self.port = port

        self.status_start_event: threading.Event = threading.Event()
        self.status_start_event.set()
        self.status_end_event: threading.Event = threading.Event()
        self.events = [self.status_start_event, self.status_end_event]

        self.http_server: MyHttpServer = MyHttpServer(
            ("localhost", self.port), HttpHandler, self
        )
        self.http_server_thread = self.http_server.serve_in_new_thread()

    def shutdown(self):
        if self.http_server.server_started_event.is_set() and not self.http_server.server_finished_event.is_set():
            self.http_server.shutdown_from_new_thread()
            self.http_server.server_finished_event.wait()

    @property
    def status(self):
        return self._status

    @status.setter
    def status(self, value):
        logger.debug(f"{self.__class__.__name__} status changed from {self._status} to {value}")
        self.events[self._status.value].clear()
        self._status = value
        self.events[self._status.value].set()

测试代码 #

def create_user_of_http_server(port) -> UserOfMyHttpServer:
    """Create and initialize UserOfMyHttpServer instance"""
    user = UserOfMyHttpServer(port)
    logger.info("Waiting for server start event")
    user.http_server.server_started_event.wait()
    logger.info("Server has started")
    return user


def shutdown_user_of_server(user_of_server: UserOfMyHttpServer):
    """Shutdown UserOfMyHttpServer and wait for completion"""
    user_of_server.shutdown()
    logger.info("Waiting for server shutdown event")
    user_of_server.http_server.server_finished_event.wait()
    logger.info("Server has finished")


def start_and_stop_server(port=1234):
    logger.info("=== Starting server cycle ===")
    user = create_user_of_http_server(port)
    shutdown_user_of_server(user)
    # 短暂等待确保资源完全释放
    time.sleep(0.2)
    logger.info("=== Server cycle completed successfully ===")


def test_start_stop_once():
    start_and_stop_server()


def test_start_stop_twice():
    start_and_stop_server()
    start_and_stop_server()


def test_start_stop_three_times():
    for i in range(3):
        start_and_stop_server()


if __name__ == "__main__":
    test_start_stop_three_times()

关键修复点说明 #

1. 彻底释放Socket资源：在MyHttpServer.shutdown()中添加self.server_close()，关闭底层监听Socket，确保端口被完全释放。
2. 优化启动等待逻辑：serve_in_new_thread中增加重试机制，循环尝试连接直到服务器响应，避免启动时的竞态条件。
3. 线程管理优化：设置线程为daemon=True，避免主线程退出后残留线程；使用join()等待线程结束，并添加超时处理。
4. 代码规范调整：重命名变量符合PEP8规范，禁用BaseHTTPRequestHandler的默认日志，统一使用自定义logger。

代码健壮性优化建议 #

• 添加端口占用检测：启动前检查目标端口是否被占用，提前抛出明确错误。
• 超时与重试机制：所有网络请求添加合理的超时和重试策略，避免单次请求失败导致流程中断。
• 资源清理的上下文管理器：为UserOfMyHttpServer实现__enter__和__exit__方法，支持with语句自动管理生命周期。
• 状态校验：在关键步骤（如启动、关闭）增加状态校验，避免重复操作（如重复关闭已停止的服务器）。
• 异常处理增强：细化异常捕获类型，针对不同错误场景给出明确的处理逻辑和日志信息。
• 单元测试覆盖：为核心功能（状态切换、服务器启停）编写单元测试，确保修改后的代码稳定性。

内容的提问来源于stack exchange，提问作者CiaranWelsh