Python Socket recv超时挂起问题排查及优化方案咨询

实现了一款仅在接收数据时执行任务的TCP服务器：当无数据可接收时，socket.recv(1024)应触发超时，捕获该异常后重新调用socket.recv(1024)。设置settimeout=10秒后，逻辑正常运行约一天，之后不再抛出超时异常。

相关代码如下：

import socket
import datetime
import sched
import time

debug = 3  # 假设debug变量已定义
def Process_received_data(data):
    pass  # 假设处理函数已实现
def Inactivate_entries():
    pass  # 假设该函数已实现

def work():
   print("Enter work at: "+str(datetime.datetime.now()))
   try:
      # read from the TCP conection
      client_socket, client_address = serversocket.accept()
      data = client_socket.recv(1024)
      if (debug>2): print('In work...data received:'+str(data))
      Process_received_data(data)   # process the received data
   except TimeoutError:
      #print('Timeout')
      a='Timeout' 
      print("Timeout at: "+str(datetime.datetime.now()))
   s.enter(0,1,work)       #schedule next call to work  

serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,1)
serversocket.bind(("192.168.1.199", 50001))
serversocket.listen(20)
serversocket.settimeout(10.0)        # wait 10 sec for a message if not - fail..

s=sched.scheduler(time.time, time.sleep)

try:
   s.enter(0,1,work)
   s.enter(43200,1,Inactivate_entries)   # run every 12 hours approx.
   s.run() 
   if ( s.empty() ) :
     print("Scheduling Q empty")
except (ConnectionResetError, UnicodeDecodeError, KeyboardInterrupt) :
   if (serversocket) :
     serversocket.close()
   print("Nice shutdown due to error at: "+str(datetime.datetime.now()))
   exit()    

运行输出显示，前期正常触发超时，但后续无超时日志输出（当前时间已到09:40）。

疑问：

• 该现象的原因是什么？
• Python的recv()是否存在内存泄漏？
• 使用的是树莓派上的Python3.12.3，此前该逻辑正常运行数周，近几日出现异常。
• 是否有更优实现方式（如使用非阻塞Socket）？

一、现象原因分析 #

1. 未关闭客户端连接导致阻塞
   代码中serversocket.accept()获取客户端连接后，仅调用client_socket.recv(1024)但未关闭client_socket。若客户端建立连接后不主动关闭、也不发送数据，client_socket.recv(1024)会无限阻塞（未给client_socket设置超时），导致work()函数卡在该步骤，无法回到serversocket.accept()的超时检测逻辑。运行一天后积累的半开连接会让work()始终阻塞在某个recv()调用上，不再打印超时日志。

2. 单线程调度器的局限性
   sched.scheduler是单线程调度器，一旦某个任务（如recv()）阻塞，整个调度器会停在该任务上，后续调度任务无法执行，自然不会再触发超时。

3. 关于内存泄漏
   Python的recv()本身不存在内存泄漏问题，问题出在未关闭的客户端连接占用系统资源（文件描述符、内存等），随着连接积累可能导致资源耗尽，但直接表现是线程阻塞而非内存泄漏。

二、优化实现方案 #

方案1：修复现有代码的连接处理问题 #

• 给客户端套接字设置超时，避免recv()无限阻塞；
• 处理完数据后强制关闭客户端连接；
• 捕获客户端套接字的超时异常，防止任务阻塞。

修改后的work()函数：

def work():
   print("Enter work at: "+str(datetime.datetime.now()))
   try:
      client_socket, client_address = serversocket.accept()
      client_socket.settimeout(10.0)  # 给客户端连接设置超时
      try:
          data = client_socket.recv(1024)
          if (debug>2): print(f'In work...data received from {client_address}:'+str(data))
          Process_received_data(data)
      except TimeoutError:
          print(f"Client recv timeout from {client_address} at: {datetime.datetime.now()}")
      finally:
          client_socket.close()  # 确保无论是否异常都关闭连接
   except TimeoutError:
      print("Timeout at: "+str(datetime.datetime.now()))
   s.enter(0,1,work)

方案2：非阻塞Socket+epoll（高并发最优方案） #

用epoll同时监听服务器套接字的连接事件和已连接客户端的可读事件，避免单连接阻塞整个服务：

import socket
import select
import datetime

def main():
    serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    serversocket.bind(("192.168.1.199", 50001))
    serversocket.listen(20)
    serversocket.setblocking(False)  # 设置为非阻塞

    epoll = select.epoll()
    epoll.register(serversocket.fileno(), select.EPOLLIN)

    connections = {}
    addresses = {}
    debug = 3

    def Process_received_data(data):
        pass

    try:
        while True:
            events = epoll.poll(10)  # 每次等待10秒，超时返回空列表
            if not events:
                print("Timeout at: "+str(datetime.datetime.now()))
                continue
            for fileno, event in events:
                if fileno == serversocket.fileno():
                    # 处理新连接
                    client_socket, client_addr = serversocket.accept()
                    client_socket.setblocking(False)
                    epoll.register(client_socket.fileno(), select.EPOLLIN)
                    connections[client_socket.fileno()] = client_socket
                    addresses[client_socket.fileno()] = client_addr
                elif event & select.EPOLLIN:
                    # 处理客户端数据
                    client_socket = connections[fileno]
                    try:
                        data = client_socket.recv(1024)
                        if data:
                            if debug>2:
                                print(f'In work...data received from {addresses[fileno]}:'+str(data))
                            Process_received_data(data)
                        else:
                            # 客户端关闭连接，清理资源
                            epoll.unregister(fileno)
                            client_socket.close()
                            del connections[fileno]
                            del addresses[fileno]
                    except Exception:
                        # 异常情况下强制清理连接
                        epoll.unregister(fileno)
                        client_socket.close()
                        del connections[fileno]
                        del addresses[fileno]
    except KeyboardInterrupt:
        pass
    finally:
        epoll.unregister(serversocket.fileno())
        epoll.close()
        serversocket.close()

if __name__ == "__main__":
    main()

该方案能高效处理多个客户端连接，不会因单个连接问题导致服务阻塞，适合长期运行。

方案3：多线程处理客户端连接 #

每个客户端连接启动独立线程处理，避免主线程阻塞：

import socket
import datetime
import threading
import sched
import time

def handle_client(client_socket, client_address):
    client_socket.settimeout(10.0)
    try:
        data = client_socket.recv(1024)
        if debug>2:
            print(f'In work...data received from {client_address}:'+str(data))
        Process_received_data(data)
    except TimeoutError:
        print(f"Client recv timeout from {client_address} at: {datetime.datetime.now()}")
    finally:
        client_socket.close()

def work():
    print("Enter work at: "+str(datetime.datetime.now()))
    try:
        client_socket, client_address = serversocket.accept()
        # 启动守护线程处理客户端，主线程退出时自动销毁
        threading.Thread(target=handle_client, args=(client_socket, client_address), daemon=True).start()
    except TimeoutError:
        print("Timeout at: "+str(datetime.datetime.now()))
    s.enter(0, 1, work)

# 其余代码与原代码一致，确保serversocket设置超时
debug = 3
def Process_received_data(data):
    pass
def Inactivate_entries():
    pass

serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,1)
serversocket.bind(("192.168.1.199", 50001))
serversocket.listen(20)
serversocket.settimeout(10.0)

s=sched.scheduler(time.time, time.sleep)

try:
   s.enter(0,1,work)
   s.enter(43200,1,Inactivate_entries)
   s.run() 
   if ( s.empty() ) :
     print("Scheduling Q empty")
except (ConnectionResetError, UnicodeDecodeError, KeyboardInterrupt) :
   if (serversocket) :
     serversocket.close()
   print("Nice shutdown due to error at: "+str(datetime.datetime.now()))
   exit()    

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