UDP案例

本文目录导读：

1. 案例一：最基础的“一发一收”通信（Python）
2. 案例二：UDP 实现多客户端群聊（线程版）
3. 案例三：UDP 实现简单的“文件传输”（带分片与序号）
4. 案例四：真实世界中的UDP应用——DNS查询
5. 案例五：UDP 的“不可靠性”演示（重传与丢包）

UDP（用户数据报协议）是网络通信中一种无连接、不可靠的传输层协议，与TCP的可靠流传输不同，UDP只负责发送数据包，不保证送达、不保证顺序，但开销低、延迟小，非常适合实时性要求高的场景（如视频通话、游戏、DNS查询）。

以下是几个由浅入深的UDP实战案例,包含核心代码和场景分析。

最基础的“一发一收”通信（Python）

场景：客户端发送“你好”,服务器接收并打印。

服务器端（udp_server.py）

import socket
# 1. 创建 UDP Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 2. 绑定本地端口（IP地址留空表示绑定所有网卡）
server_address = ('0.0.0.0', 8888)
server_socket.bind(server_address)
print('服务器已启动，等待数据...')
# 3. 循环接收数据
while True:
    data, client_addr = server_socket.recvfrom(1024)  # 一次最多接收1024字节
    print(f'收到来自 {client_addr} 的消息：{data.decode()}')
    # 可选：回复客户端
    reply = f'已收到你的消息'.encode()
    server_socket.sendto(reply, client_addr)

客户端（udp_client.py）

import socket
# 1. 创建 UDP Socket（不需要连接）
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 2. 发送数据给服务器（不需要 connect）
server_addr = ('127.0.0.1', 8888)
message = '你好，服务器！'.encode()
client_socket.sendto(message, server_addr)
# 3. 等待服务器回复（非必须，可设置超时防止卡死）
client_socket.settimeout(3)  # 3秒收不到回复则报错
try:
    reply, server_addr = client_socket.recvfrom(1024)
    print(f'服务器回复：{reply.decode()}')
except socket.timeout:
    print('等待超时，服务器未回复。')

UDP 实现多客户端群聊（线程版）

场景：多个客户端连接到同一个服务器，任意客户端发送的消息会广播给所有人。 关键点：UDP服务器使用字典记录所有客户端地址,收到消息后遍历转发。

import socket
import threading
# 全局变量：存储所有连接的客户端地址（IP, Port）
clients = set()
lock = threading.Lock()
def broadcast(message, sender_addr):
    """广播消息给所有其他客户端"""
    with lock:
        for client in clients:
            if client != sender_addr:  # 不发给发送者自己
                server_socket.sendto(message, client)
# 创建服务器 Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind(('0.0.0.0', 9999))
print('群聊服务器启动...')
# 处理接收消息的函数（放在线程中运行）
def recv_handler():
    while True:
        data, addr = server_socket.recvfrom(1024)
        # 将新地址添加到客户端列表
        with lock:
            if addr not in clients:
                clients.add(addr)
                print(f'新用户加入：{addr}')
        # 转发消息
        broadcast(data, addr)
# 启动接收线程（主线程等待）
thread = threading.Thread(target=recv_handler)
thread.daemon = True
thread.start()
# 主线程保持运行（例如按Ctrl+C退出）
try:
    while True:
        pass
except KeyboardInterrupt:
    print('服务器关闭')

客户端只需创建socket后，通过sendto发送，并循环recvfrom接收广播即可。

UDP 实现简单的“文件传输”（带分片与序号）

场景：发送大文件时，由于UDP单包上限（MTU通常1500字节），需要将文件分片发送，并加上序号方便重组。 注意：此案例不实现重传（那是RUDP/QUIC做的事）,仅演示分片逻辑。

import socket
import os
def send_file(file_path, server_addr):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    file_size = os.path.getsize(file_path)
    # 1. 先发送文件信息（文件名长度+文件名+文件大小）
    file_name = os.path.basename(file_path).encode()
    meta = f'{len(file_name)}:'.encode() + file_name + f':{file_size}'.encode()
    s.sendto(b'META|' + meta, server_addr)
    # 2. 分片发送内容（每片 1400 字节，留出头部空间）
    seq = 0
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(1400)
            if not chunk:
                break
            # 数据包格式：[序号(4字节)][数据]
            packet = seq.to_bytes(4, 'big') + chunk
            s.sendto(packet, server_addr)
            seq += 1
    # 3. 发送结束标志
    s.sendto(b'END|' + seq.to_bytes(4,'big'), server_addr)
    print(f'文件发送完成，共 {seq} 个分片')
# 接收端（伪代码）
def recv_file(server_socket):
    # 1. 接收META，解析文件名和大小
    # 2. 循环接收分片，根据序号存入字典
    # 3. 收到END后，按顺序写文件
    pass

真实世界中的UDP应用——DNS查询

场景：通过UDP直接向DNS服务器（如8.8.8.8）发送查询报文，获取域名IP。 特点：UDP自带“请求-响应”模式,非常适合这种短小查询。

import socket
import struct
# 1. 构造DNS查询报文（精简版）
def build_dns_query(domain):
    header = struct.pack('>HHHHHH', 0x1234, 0x0100, 1, 0, 0, 0)  # 事务ID, 标志, 问题数...
    qname = b''.join(bytes([len(part)]) + part.encode() for part in domain.split('.')) + b'\x00'
    qtype = struct.pack('>HH', 1, 1)  # A记录, IN类
    return header + qname + qtype
# 2. 发送并接收
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(5)
dns_server = ('8.8.8.8', 53)
query = build_dns_query('www.example.com')
s.sendto(query, dns_server)
try:
    response, _ = s.recvfrom(512)
    # 解析响应中的IP地址（简化处理，只找第一个A记录）
    current_index = 12  # 跳过头部
    # ...复杂解析省略...
    print('DNS响应已接收')
except socket.timeout:
    print('DNS查询超时')

UDP 的“不可靠性”演示（重传与丢包）

目的：展示UDP不保证送达,需要上层自己处理丢包。

import socket
import random
import time
# 模拟一个丢包率为50%的服务器
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind(('0.0.0.0', 7777))
print('丢包服务器启动（50%概率丢弃消息）')
while True:
    data, addr = server_socket.recvfrom(1024)
    if random.random() < 0.5:
        print(f'丢弃来自 {addr} 的数据包')
        continue  # 直接不回复
    reply = b'OK'
    server_socket.sendto(reply, addr)
# 客户端需要设置超时并重试
client_socket.settimeout(1)
for attempt in range(5):
    client_socket.sendto(b'ping', server_addr)
    try:
        reply = client_socket.recvfrom(1024)
        print('成功收到回复')
        break
    except socket.timeout:
        print(f'第 {attempt+1} 次尝试超时，重试...')

编程提示：

• 使用 sendto() 发送时必须指定目标地址。
• 使用 recvfrom() 接收时会返回数据包大小限制内的内容，一次接收一个完整UDP报文,不会像TCP那样拆包粘包。
• 如果需要可靠传输，你必须自己在应用层实现：确认机制（ACK）、超时重传、序列号、流量控制（这就是RUDP/QUIC的原理）。

案例涵盖了UDP的基础用法、广播实现、分片传输和实际网络协议应用，初学者建议从案例一开始动手,逐步加深理解。