如何退出Python的TCP accept函数？实现超时后自动退出

方法1：给监听socket设置超时时间 #

通过settimeout()给监听socket指定超时时长，当accept()在设定时间内未接收到连接时，会抛出socket.timeout异常，捕获该异常即可让程序正常退出。

修改后的代码：

import socket

HOST = "127.0.0.1"
PORT = 65432

def connect():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((HOST, PORT))
        print("listening")
        s.listen()
        # 设置超时时间为10秒
        s.settimeout(10)
        print("accepting，10秒内无连接则退出")
        try:
            conn, addr = s.accept()
            with conn:
                print(f"Connected by {addr}")
                while True:
                    data = conn.recv(1024)
                    if not data:
                        break
                    conn.sendall(data)
        except socket.timeout:
            print("超时，程序退出")

def main():
    connect()

if __name__ == "__main__":
    main()

方法2：使用select.select监听连接 #

利用select.select监听监听socket的可读状态，同时设置超时参数，超时后直接终止程序。这种方式更灵活，适合需要同时处理多个socket的场景。

修改后的代码：

import socket
import select

HOST = "127.0.0.1"
PORT = 65432

def connect():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((HOST, PORT))
        print("listening")
        s.listen()
        print("accepting，10秒内无连接则退出")
        # 监听可读事件，超时10秒
        readable, _, _ = select.select([s], [], [], 10)
        if not readable:
            print("超时，程序退出")
            return
        # 有可读事件，说明有连接请求
        conn, addr = s.accept()
        with conn:
            print(f"Connected by {addr}")
            while True:
                data = conn.recv(1024)
                if not data:
                    break
                conn.sendall(data)

def main():
    connect()

if __name__ == "__main__":
    main()

方法3：线程+定时器强制关闭socket #

启动定时器线程，超时后关闭监听socket，此时accept()会触发OSError异常，捕获异常后即可退出程序。这种方式适合需要强制终止阻塞操作的场景。

修改后的代码：

import socket
import threading

HOST = "127.0.0.1"
PORT = 65432

def connect():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((HOST, PORT))
        print("listening")
        s.listen()
        print("accepting，10秒内无连接则退出")
        
        # 定义超时后关闭socket的函数
        def timeout_close():
            print("超时，关闭监听socket")
            s.close()
        
        # 启动10秒定时器
        timer = threading.Timer(10, timeout_close)
        timer.start()
        
        try:
            conn, addr = s.accept()
            # 有连接接入，取消定时器
            timer.cancel()
            with conn:
                print(f"Connected by {addr}")
                while True:
                    data = conn.recv(1024)
                    if not data:
                        break
                    conn.sendall(data)
        except OSError:
            # socket被关闭，捕获异常退出
            print("程序退出")

def main():
    connect()

if __name__ == "__main__":
    main()

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