UDP 特性
- 無連線、不可靠、無序、無流控
- 適合:即時影音、遊戲狀態同步、DNS 查詢、廣播/多播
- 不適合:檔案傳輸、重要指令、需保證送達的資料
伺服器端
import socket
def udp_server(host='', port=8081, buffer_size=1024):
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((host, port))
print(f"UDP 伺服器監聽 {host}:{port}")
while True:
data, addr = sock.recvfrom(buffer_size)
print(f"收自 {addr}: {data.decode('utf-8', errors='replace')}")
# 回覆
sock.sendto(b"ACK: " + data, addr)
if __name__ == '__main__':
udp_server()
客戶端
import socket
def udp_client(server_host='localhost', server_port=8081):
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(5.0) # 避免永久阻塞
try:
message = "Hello UDP Server!"
sock.sendto(message.encode('utf-8'), (server_host, server_port))
print(f"已送出: {message}")
data, addr = sock.recvfrom(1024)
print(f"收到回覆 {addr}: {data.decode()}")
except socket.timeout:
print("等待回覆逾時")
finally:
sock.close()
if __name__ == '__main__':
udp_client()
進階:非阻塞 + select 多工
import socket, select
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('', 8081))
sock.setblocking(False)
while True:
ready, _, _ = select.select([sock], [], [], 1.0)
if ready:
data, addr = sock.recvfrom(1024)
print(f"{addr}: {data.decode()}")
# 可在此處理其他邏輯
常見問題
| 問題 | 解決 |
| 資料截斷 | 增大 buffer_size(最大 65507 位元組) |
| 封包遺失 | 應用層實作 ACK/重傳/序號(或改用 TCP) |
| 防火牆阻擋 | 開放 UDP 埠、允許 ICMP Port Unreachable |
| NAT 穿透 | 使用 STUN/TURN 或雙方同時 sendto 打洞 |