从外部函数关闭 Python Socket 服务器

import socket
import threading

def create_server(address: str, port: int, close_event: threading.Event):
    """
    创建一个 Socket 服务器，并在单独的线程中运行。

    Args:
        address: 服务器绑定的地址。
        port: 服务器监听的端口。
        close_event: 一个线程事件，用于通知服务器关闭。
    """
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind((address, port))
    s.listen(5)
    print(f"Server listening on {address}:{port}")
    while not close_event.is_set():
        try:
            conn, addr = s.accept()
            print(f"Connected to {addr}!")
            while conn:
                data = conn.recv(1024).decode()
                if data == "end":
                    print(f"Closing connection from {addr}")
                    conn.close()
                    break # Exit the inner loop for this connection
                elif data:
                    print(f"Received: {data} from {addr}")
                    # Process the received data here
                else:
                    # Connection closed by client
                    print(f"Connection closed by {addr}")
                    conn.close()
                    break # Exit the inner loop for this connection
        except socket.timeout:
            # Check the event every timeout
            pass
        except Exception as e:
            print(f"Error in server loop: {e}")
            break # Exit the outer loop if an error occurs

    print("Server shutting down...")
    try:
        s.shutdown(socket.SHUT_RDWR)
    except OSError as e:
        print(f"Error during shutdown: {e}")  # Handle cases where the socket might already be closed
    finally:
        s.close()
    print("Server stopped.")


def close_socket(close_event: threading.Event):
    """
    设置线程事件，通知 Socket 服务器关闭。

    Args:
        close_event: 要设置的线程事件。
    """
    print("Requesting server shutdown...")
    close_event.set()


# Example Usage:
if __name__ == "__main__":
    server_close_event = threading.Event()
    # Set a timeout for the socket, so it doesn't block indefinitely on accept()
    # This allows it to check the close_event regularly
    address = "127.0.0.1"  # Loopback address for testing
    port = 12345
    server_thread = threading.Thread(target=create_server, args=(address, port, server_close_event))
    server_thread.daemon = True # Allow the main program to exit even if the thread is still running
    server_thread.start()

    # Simulate some work in the main thread
    import time
    time.sleep(5)

    # Stop the server
    close_socket(server_close_event)
    server_thread.join() # Wait for the server thread to finish

    print("Program finished.")