如何检测本地localhost运行的Python UDP服务是否可用

UDP是无连接传输协议，不存在TCP那样的三次握手连接流程：

• set_dest_address() 仅在本地套接字上配置目标地址和端口的转发规则，不会和对端做任何网络交互，只要传入的IP、端口格式合法就会返回0，根本不会校验对端是否真的有服务在监听。
• put_packet() 执行时仅把数据包交给本地操作系统网络栈发送，不会等待对端确认，哪怕对端没启动，只要本地网络栈正常就不会抛出错误。

操作系统虽然会在收到发往未监听UDP端口的数据包时返回ICMP端口不可达报文，但这个反馈受防火墙、NAT网络环境影响极大，且Godot的UDP接口默认不会直接把这个状态暴露给上层，无法作为可靠的判断依据。

UDP本身没有连接状态的概念，连通性校验必须在应用层实现，核心逻辑是：客户端发送探测包，服务端收到后返回约定的应答包，客户端在超时时间内收到应答就判定服务可用，否则判定不可达。

Godot端实现代码 #

var udp_socket = PacketPeerUDP.new()
# 绑定本地随机空闲端口，用于接收服务端返回的应答
var bind_err = udp_socket.bind(0, "0.0.0.0")
if bind_err != 0:
    print("UDP套接字初始化失败，错误码：", bind_err)
    return
udp_socket.set_dest_address("127.0.0.1", 6000)

var is_service_online = false
var retry_count = 3 # 重试3次降低网络波动导致的误判
var timeout_ms = 300 # 每次等待应答超时300ms

for i in range(retry_count):
    # 发送心跳探测包
    var heartbeat = JSON.print({
        "msg_type": "probe",
        "timestamp": Time.get_ticks_msec()
    }).to_ascii()
    udp_socket.put_packet(heartbeat)

    # 等待应答
    var start_time = Time.get_ticks_msec()
    while Time.get_ticks_msec() - start_time < timeout_ms:
        udp_socket.poll() # 必须轮询刷新套接字缓冲区才能收到数据
        if udp_socket.get_available_packet_count() > 0:
            var resp_raw = udp_socket.get_packet().get_string_from_ascii()
            var resp = JSON.parse_string(resp_raw)
            if resp and resp.get("msg_type") == "probe_ack":
                is_service_online = true
                break
        await get_tree().process_frame
    if is_service_online:
        break

if not is_service_online:
    print("目标UDP服务不可用，终止数据发送")
else:
    print("UDP服务连通正常，开始传输业务数据")
    # 此处写正常发送业务数据的逻辑

Python服务端适配代码 #

import socket
import json

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(("127.0.0.1", 6000))
sock.settimeout(0.1) # 设置非阻塞超时，避免循环卡死

while True:
    try:
        data, client_addr = sock.recvfrom(2048)
        try:
            pkt = json.loads(data.decode("utf-8"))
            # 收到探测包直接返回应答
            if pkt.get("msg_type") == "probe":
                ack = json.dumps({"msg_type": "probe_ack"}).encode("utf-8")
                sock.sendto(ack, client_addr)
                continue
            # 以下为正常业务数据处理逻辑
            print("收到业务数据：", pkt)
        except json.JSONDecodeError:
            print("收到非JSON格式的非法数据包")
    except socket.timeout:
        continue

可以尝试捕获操作系统返回的ICMP不可达错误：发送数据包后多次调用poll()刷新套接字，再调用udp_socket.get_error()获取错误码，若返回连接重置/连接拒绝类错误（Windows错误码10054、Linux错误码111），可判定端口未监听。
注意：该方案兼容性极差，系统防火墙、跨网段NAT都会拦截ICMP报文导致误判，只能作为辅助判断手段，不能作为唯一依据。

• 业务传输过程中也可以每隔几秒发送一次心跳，及时检测服务中途掉线的情况
• 如果业务需要强可靠传输、自动重传、明确的连接状态，不要用UDP，直接换TCP实现。

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