Python Socket技术问题：检查连接状态、收发就绪及服务器消息

你提供的UDP连接代码如下：

import socket

# 假设self.get_free_port()是自定义获取空闲端口的方法
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
bind = '0.0.0.0'
lport = self.get_free_port()
host = (str(ip), int(rport))
sock.settimeout(15)
sock.connect(host)

针对你提出的三个技术问题，解决方案如下：

1. 如何检查socket是否已准备好发送和接收数据？ #

UDP是无连接协议，但调用connect()后会固定目标地址。可以用select模块做非阻塞式检查，无需实际收发数据：

import select

# 检查是否可写（准备好发送）
write_ready, _, _ = select.select([], [sock], [], 0)
if write_ready:
    print("Socket已准备好发送数据")

# 检查是否可读（准备好接收）
read_ready, _, _ = select.select([sock], [], [], 0)
if read_ready:
    print("Socket已准备好接收数据")

select.select()最后一个参数设为0表示立即返回结果，不会阻塞程序。

2. 如何检查服务器是否向我发送了hello消息？ #

可以通过带超时的接收逻辑检测，避免程序长期阻塞；也可以将socket设为非阻塞模式实现轮询：

try:
    # 尝试接收最多1024字节数据
    data, addr = sock.recvfrom(1024)
    if data.decode().strip() == "hello":
        print("收到服务器的hello消息")
    else:
        print(f"收到其他消息: {data.decode()}")
except socket.timeout:
    print("超时未收到hello消息")
except BlockingIOError:
    # socket设为非阻塞时，无数据会抛出该异常
    print("当前没有收到任何消息")

若要开启非阻塞模式，执行以下代码：

sock.setblocking(False)

3. 如何确认当前socket连接是否正常？ #

UDP本身没有连接状态，connect()仅固定了目标地址，无法像TCP那样直接检测存活状态。常用方案是发送心跳包并等待响应：

def check_socket_alive(sock, target_host):
    try:
        # 发送心跳包
        sock.sendall(b"ping")
        # 等待响应
        data, addr = sock.recvfrom(1024)
        return addr == target_host and data.decode().strip() == "pong"
    except socket.timeout:
        return False
    except Exception as e:
        print(f"检查连接时出错: {e}")
        return False

# 调用检查逻辑
is_alive = check_socket_alive(sock, host)
print("连接正常" if is_alive else "连接异常")

如果服务器没有心跳响应机制，也可通过检测发送时的系统错误判断：

import errno
import os

try:
    sock.sendall(b"test")
    # 获取socket错误状态
    error_code = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
    if error_code != 0:
        raise OSError(error_code, os.strerror(error_code))
except OSError as e:
    if e.errno == errno.ECONNREFUSED:
        print("服务器拒绝连接")
    elif e.errno == errno.EHOSTUNREACH:
        print("目标主机不可达")
    else:
        print(f"连接异常: {e}")

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