Python Socket：客户端心跳断开检测与单用户连接数限制实现咨询

需求 #

• 服务端接收客户端心跳，若客户端数分钟未响应则关闭其Socket
• 统计单用户连接数，最大限制为5个

当前困惑 #

• 客户端持续发送心跳，但服务端的recv方法处于阻塞状态，无法执行心跳超时检测
• 不确定当前通过增减全局计数器统计用户连接数的方法是否合理

现有代码 #

服务端代码 #

import socket 
from _thread import *
import sys
import time


username = ['test123', 'hongengi']
userconnect= 0


def threaded(client_socket, addr): 
    print('Connected by :', addr[0], ':', addr[1]) 


    while True: 

        try:

            data = client_socket.recv(1024)
            print (data.decode())
            print('Received from ' + addr[0],':',addr[1] , data.decode())

            if data.decode() == ".": # heartbeat
                heartbeat = time.perf_counter()
                print ("heartbeat")

            if data.decode() == "test123":
                print ("login success")
                userconnect == userconnect + 1    
                
            
            if not data:
                print ("no data / disconnect ")
                print('Disconnected by ' + addr[0],':',addr[1])
                userconnect == userconnect - 1 
                break

            client_socket.send(data) 

        except (ConnectionResetError, socket.error) as e:
            print ("error occurs")

            print('Disconnected by ' + addr[0],':',addr[1])
            userconnect == userconnect - 1 
            break
             
    client_socket.close() 


HOST = '127.0.0.1'
PORT = 5000

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind((HOST, PORT)) 
server_socket.listen() 

print('server start')


while True: 

    print('wait')


    client_socket, addr = server_socket.accept() 
    start_new_thread(threaded, (client_socket, addr)) 

server_socket.close()

客户端代码 #

import socket

SERVER_IP = 'localhost'
SERVER_PORT = 5000
SIZE = 100
SERVER_ADDR = (SERVER_IP, SERVER_PORT)
heartbeat = "."
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(SERVER_ADDR)
#username = "test123"
#userpass = "123123"

while True:

    client_socket.send(heartbeat.encode())
    msg = client_socket.recv(SIZE)
    print (msg.decode())

解决方案 #

1. 解决阻塞recv的心跳超时检测问题 #

阻塞式recv会导致无法主动检查心跳时间，可通过以下方式处理：

为Socket设置超时时间

在客户端连接后，给client_socket设置超时时间，当recv触发超时异常时，检查该客户端的最后心跳时间是否超过阈值，若超过则关闭连接：

def threaded(client_socket, addr): 
    print('Connected by :', addr[0], ':', addr[1]) 
    # 设置recv超时为30秒（可按需调整）
    client_socket.settimeout(30)
    last_heartbeat = time.perf_counter()
    # 心跳超时阈值，比如5分钟
    HEARTBEAT_TIMEOUT = 300

    while True: 
        try:
            data = client_socket.recv(1024)
            if data:
                data_str = data.decode()
                print('Received from ' + addr[0],':',addr[1] , data_str)

                if data_str == ".": # 心跳包
                    last_heartbeat = time.perf_counter()
                    print ("heartbeat received")
                elif data_str == "test123":
                    print ("login success")
                    # 后续处理用户登录逻辑

                client_socket.send(data) 
            else:
                # 客户端主动断开
                print ("no data / disconnect ")
                print('Disconnected by ' + addr[0],':',addr[1])
                break

        except socket.timeout:
            # 超时触发，检查心跳是否过期
            if time.perf_counter() - last_heartbeat > HEARTBEAT_TIMEOUT:
                print(f"Client {addr} heartbeat timeout, closing connection")
                break
            # 未过期则继续循环等待
            continue
        except (ConnectionResetError, socket.error) as e:
            print ("error occurs:", e)
            print('Disconnected by ' + addr[0],':',addr[1])
            break
             
    client_socket.close() 

2. 单用户连接数统计的合理方案 #

现有代码存在两个核心问题：

• 使用==进行赋值操作，属于语法错误，应改为+=和-=
• 全局计数器userconnect无法区分不同用户，无法实现单用户的连接数限制

正确方案是使用字典按用户名统计连接数，同时添加线程锁保证并发安全：

import socket 
from _thread import *
import sys
import time
from collections import defaultdict
import threading

username = ['test123', 'hongengi']
# 按用户名存储当前连接数，默认值为0
user_connection_count = defaultdict(int)
# 单用户最大连接数限制
MAX_CONNECTIONS_PER_USER = 5
# 线程锁，保护连接数字典的并发修改
count_lock = threading.Lock()


def threaded(client_socket, addr): 
    print('Connected by :', addr[0], ':', addr[1]) 
    client_socket.settimeout(30)
    last_heartbeat = time.perf_counter()
    HEARTBEAT_TIMEOUT = 300
    current_user = None  # 记录当前连接对应的用户名

    while True: 
        try:
            data = client_socket.recv(1024)
            if data:
                data_str = data.decode()
                print('Received from ' + addr[0],':',addr[1] , data_str)

                if data_str == ".":
                    last_heartbeat = time.perf_counter()
                    print ("heartbeat received")
                elif data_str in username:  # 假设收到用户名即完成登录
                    current_user = data_str
                    # 加锁检查并修改连接数
                    with count_lock:
                        if user_connection_count[current_user] >= MAX_CONNECTIONS_PER_USER:
                            client_socket.send(b"Connection limit exceeded")
                            print(f"User {current_user} reached max connections, closing")
                            break
                        user_connection_count[current_user] += 1
                    print ("login success")
                    client_socket.send(b"Login success")

                else:
                    client_socket.send(data) 
            else:
                print ("no data / disconnect ")
                print('Disconnected by ' + addr[0],':',addr[1])
                break

        except socket.timeout:
            if time.perf_counter() - last_heartbeat > HEARTBEAT_TIMEOUT:
                print(f"Client {addr} heartbeat timeout, closing connection")
                break
            continue
        except (ConnectionResetError, socket.error) as e:
            print ("error occurs:", e)
            print('Disconnected by ' + addr[0],':',addr[1])
            break
             
    # 断开连接时，若已登录则减少对应用户的连接数
    if current_user:
        with count_lock:
            user_connection_count[current_user] -= 1
            # 避免出现负数（异常场景下）
            if user_connection_count[current_user] < 0:
                user_connection_count[current_user] = 0
    client_socket.close() 


HOST = '127.0.0.1'
PORT = 5000

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind((HOST, PORT)) 
server_socket.listen() 

print('server start')


while True: 
    print('wait')
    client_socket, addr = server_socket.accept() 
    start_new_thread(threaded, (client_socket, addr)) 

server_socket.close()

该方案优势：

• 精准统计每个用户的连接数，实现单用户连接限制
• 通过线程锁避免多线程并发修改连接数字典导致的计数错误

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