首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
您现在的位置: 程式碼之家 >> 編程語言 >  >> 更多程式語言 >> 更多程式設計知識

Python UDP 網路程式設計:Socket 資料傳送與接收完整實作

     編輯:更多程式設計知識

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 打洞
  1. 上一篇文章:
  2. 下一篇文章:
Copyright © 程式碼之家 All Rights Reserved