Python實現(xiàn)基于UDP的文件傳輸?shù)娜^程
更新時間：2025年12月10日 10:11:22 作者：剛哥的進化路
本文介紹了UDP文件傳輸協(xié)議的設(shè)計思路,包括分塊傳輸、包標識、確認重傳等關(guān)鍵步驟,通過客戶端和服務(wù)端代碼實現(xiàn)文件傳輸,并提供了基礎(chǔ)版使用步驟、關(guān)鍵優(yōu)化和避坑指南,以確保文件傳輸?shù)姆€(wěn)定性和可靠性,需要的朋友可以參考下
UDP本身是無連接、不可靠的傳輸協(xié)議，直接用來傳文件容易出現(xiàn)丟包、亂序問題，但通過“分塊傳輸+校驗+重傳”的設(shè)計，依然能實現(xiàn)穩(wěn)定的文件傳輸。
一、UDP文件傳輸?shù)暮诵脑O(shè)計思路
UDP不保證數(shù)據(jù)送達，所以要先解決三個關(guān)鍵問題：
- 分塊傳輸：將大文件拆分成固定大小的數(shù)據(jù)包（如1024字節(jié)），避免單次發(fā)送數(shù)據(jù)過大導致丟包；
- 包標識：每個數(shù)據(jù)包添加“序號+總包數(shù)+校驗位”，服務(wù)端接收后能校驗完整性、排序重組；
- 確認重傳：服務(wù)端接收每個數(shù)據(jù)包后返回確認（ACK），客戶端未收到ACK則重傳該數(shù)據(jù)包，保證可靠性。
核心流程：
- 客戶端：讀取文件→分塊→加標識→發(fā)送數(shù)據(jù)包→等待ACK→重傳失敗包→發(fā)送結(jié)束標識；
- 服務(wù)端：接收數(shù)據(jù)包→校驗→排序→重組→保存文件→返回ACK。
二、完整實現(xiàn)代碼
1. 服務(wù)端代碼（接收文件）
import socket
import os
# 配置參數(shù)
UDP_IP = "" # 監(jiān)聽所有本機IP
UDP_PORT = 9999
BUFFER_SIZE = 1024 # 數(shù)據(jù)包大?。ㄐ韬涂蛻舳艘恢拢?
SAVE_DIR = "received_files" # 文件保存目錄
# 創(chuàng)建保存目錄
if not os.path.exists(SAVE_DIR):
os.makedirs(SAVE_DIR)
def udp_file_server():
# 1. 創(chuàng)建UDP Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind((UDP_IP, UDP_PORT))
# 允許端口復用，避免重啟報錯
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
print(f"UDP文件服務(wù)端已啟動，監(jiān)聽端口 {UDP_PORT}...")
# 初始化接收狀態(tài)
file_data = {} # 存儲接收的數(shù)據(jù)包 {序號: 數(shù)據(jù)}
total_packets = 0 # 總包數(shù)
file_name = "" # 文件名
client_addr = None # 客戶端地址
try:
while True:
# 2. 接收數(shù)據(jù)包（阻塞）
data, client_addr = server_socket.recvfrom(BUFFER_SIZE + 32) # 預留標識位空間
if not data:
continue
# 解析數(shù)據(jù)包標識："文件名|總包數(shù)|當前序號|數(shù)據(jù)"
try:
# 分割標識和數(shù)據(jù)（用|分隔，最后一段是文件數(shù)據(jù)）
parts = data.decode('utf-8', errors='ignore').split('|', 3)
if len(parts) != 4:
# 不是文件數(shù)據(jù)包，可能是結(jié)束標識
if data.decode('utf-8') == "TRANSFER_FINISH":
print("客戶端發(fā)送結(jié)束標識，開始重組文件...")
break
continue
file_name, total_packets_str, packet_num_str, file_chunk = parts
total_packets = int(total_packets_str)
packet_num = int(packet_num_str)
except Exception as e:
print(f"數(shù)據(jù)包解析失?。簕e}")
# 返回錯誤ACK
server_socket.sendto(f"ERR|{packet_num}".encode('utf-8'), client_addr)
continue
# 3. 校驗并保存數(shù)據(jù)包
if 1 <= packet_num <= total_packets:
file_data[packet_num] = file_chunk
# 返回成功ACK
server_socket.sendto(f"ACK|{packet_num}".encode('utf-8'), client_addr)
# 打印進度
progress = (len(file_data) / total_packets) * 100
print(f"接收進度：{progress:.1f}% ({len(file_data)}/{total_packets})", end='\r')
# 4. 重組并保存文件
if file_data and total_packets > 0:
# 按序號排序數(shù)據(jù)包
sorted_chunks = [file_data[i] for i in range(1, total_packets + 1) if i in file_data]
# 拼接所有數(shù)據(jù)
file_path = os.path.join(SAVE_DIR, file_name)
with open(file_path, 'wb') as f:
for chunk in sorted_chunks:
f.write(chunk.encode('utf-8')) # 若傳二進制文件，需調(diào)整編碼邏輯（見進階部分）
print(f"\n文件接收完成！保存路徑：{file_path}")
# 發(fā)送完成確認
server_socket.sendto("FILE_SAVED".encode('utf-8'), client_addr)
else:
print("\n未接收到完整文件數(shù)據(jù)")
except KeyboardInterrupt:
print("\n服務(wù)端手動停止")
finally:
server_socket.close()
if __name__ == "__main__":
udp_file_server()
2. 客戶端代碼（發(fā)送文件）
import socket
import os
import time
# 配置參數(shù)
SERVER_IP = "127.0.0.1" # 服務(wù)端IP（遠程傳輸改實際IP）
SERVER_PORT = 9999
BUFFER_SIZE = 1024 # 每個數(shù)據(jù)包的文件數(shù)據(jù)大小
RETRY_TIMES = 3 # 單個數(shù)據(jù)包最大重傳次數(shù)
RETRY_INTERVAL = 0.5 # 重傳間隔（秒）
def udp_file_client(file_path):
# 1. 檢查文件是否存在
if not os.path.exists(file_path):
print(f"錯誤：文件 {file_path} 不存在")
return
# 獲取文件名和文件大小
file_name = os.path.basename(file_path)
file_size = os.path.getsize(file_path)
# 計算總包數(shù)（向上取整）
total_packets = (file_size + BUFFER_SIZE - 1) // BUFFER_SIZE
print(f"準備發(fā)送文件：{file_name}，大?。簕file_size}字節(jié)，總包數(shù)：{total_packets}")
# 2. 創(chuàng)建UDP Socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
client_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 設(shè)置超時（避免等待ACK卡死）
client_socket.settimeout(2)
try:
# 3. 讀取文件并分塊發(fā)送
with open(file_path, 'r', encoding='utf-8') as f: # 二進制文件用'rb'（見進階部分）
for packet_num in range(1, total_packets + 1):
# 讀取當前塊數(shù)據(jù)
file_chunk = f.read(BUFFER_SIZE)
if not file_chunk:
break
# 構(gòu)造數(shù)據(jù)包："文件名|總包數(shù)|當前序號|數(shù)據(jù)"
packet_data = f"{file_name}|{total_packets}|{packet_num}|{file_chunk}".encode('utf-8')
retry_count = 0
ack_received = False
# 4. 發(fā)送并等待ACK，失敗則重傳
while retry_count < RETRY_TIMES and not ack_received:
try:
# 發(fā)送數(shù)據(jù)包
client_socket.sendto(packet_data, (SERVER_IP, SERVER_PORT))
# 等待ACK
ack_data, _ = client_socket.recvfrom(128)
ack_parts = ack_data.decode('utf-8').split('|')
if ack_parts[0] == "ACK" and int(ack_parts[1]) == packet_num:
ack_received = True
# 打印進度
progress = (packet_num / total_packets) * 100
print(f"發(fā)送進度：{progress:.1f}% ({packet_num}/{total_packets})", end='\r')
except socket.timeout:
retry_count += 1
print(f"\n數(shù)據(jù)包 {packet_num} 超時，重傳 {retry_count}/{RETRY_TIMES}")
time.sleep(RETRY_INTERVAL)
except Exception as e:
print(f"\n數(shù)據(jù)包 {packet_num} 發(fā)送失?。簕e}")
retry_count += 1
time.sleep(RETRY_INTERVAL)
if not ack_received:
print(f"\n錯誤：數(shù)據(jù)包 {packet_num} 重傳{RETRY_TIMES}次失敗，傳輸終止")
return
# 5. 發(fā)送結(jié)束標識
client_socket.sendto("TRANSFER_FINISH".encode('utf-8'), (SERVER_IP, SERVER_PORT))
# 等待服務(wù)端保存完成確認
try:
finish_ack, _ = client_socket.recvfrom(128)
if finish_ack.decode('utf-8') == "FILE_SAVED":
print("\n文件發(fā)送完成！服務(wù)端已保存")
except socket.timeout:
print("\n未收到服務(wù)端完成確認，但數(shù)據(jù)已發(fā)送完畢")
except KeyboardInterrupt:
print("\n客戶端手動停止")
finally:
client_socket.close()
if __name__ == "__main__":
# 替換為你要發(fā)送的文件路徑（本地測試用絕對/相對路徑）
target_file = "test.txt" # 示例：發(fā)送當前目錄的test.txt
udp_file_client(target_file)
三、基礎(chǔ)版使用步驟
- 準備測試文件：在客戶端目錄創(chuàng)建一個
test.txt文件（內(nèi)容任意）； - 啟動服務(wù)端：運行服務(wù)端代碼，控制臺顯示“UDP文件服務(wù)端已啟動”；
- 啟動客戶端：修改客戶端代碼中
target_file為實際文件路徑，運行客戶端； - 查看結(jié)果：服務(wù)端控制臺顯示傳輸進度，完成后文件會保存到
received_files目錄。
四、關(guān)鍵優(yōu)化：支持二進制文件（圖片/視頻/壓縮包）
基礎(chǔ)版僅支持文本文件，要傳輸圖片、視頻等二進制文件，需修改編碼邏輯（核心是避免字符編碼導致的數(shù)據(jù)損壞）：
1. 客戶端修改（讀取二進制文件）
# 替換客戶端文件讀取部分
with open(file_path, 'rb') as f: # 改為二進制讀取
for packet_num in range(1, total_packets + 1):
file_chunk = f.read(BUFFER_SIZE)
if not file_chunk:
break
# 構(gòu)造數(shù)據(jù)包：用特殊分隔符（如b'|||'），避免二進制數(shù)據(jù)沖突
packet_header = f"{file_name}|{total_packets}|{packet_num}".encode('utf-8')
packet_data = packet_header + b'|||' + file_chunk # 二進制拼接
2. 服務(wù)端修改（解析二進制數(shù)據(jù)）
# 替換服務(wù)端數(shù)據(jù)包解析部分
try:
# 分割頭部和二進制數(shù)據(jù)（按b'|||'分割）
header, file_chunk = data.split(b'|||', 1)
# 解析頭部（轉(zhuǎn)字符串）
parts = header.decode('utf-8').split('|', 2)
file_name, total_packets_str, packet_num_str = parts
total_packets = int(total_packets_str)
packet_num = int(packet_num_str)
except Exception as e:
print(f"二進制數(shù)據(jù)包解析失敗：{e}")
continue
# 保存時直接寫入二進制數(shù)據(jù)
with open(file_path, 'wb') as f:
for chunk in sorted_chunks:
f.write(chunk) # 無需encode，直接寫二進制
五、避坑指南：常見問題與解決方案
1. 數(shù)據(jù)包丟包/重傳失敗
- 原因：UDP無可靠性保證，網(wǎng)絡(luò)波動易丟包；
- 解決方案：
- 增大
RETRY_TIMES（如改為5），延長RETRY_INTERVAL； - 減小
BUFFER_SIZE（如改為512），降低單包傳輸壓力； - 遠程傳輸時確保服務(wù)端端口已開放防火墻。
- 增大
2. 大文件傳輸卡頓
- 原因：循環(huán)發(fā)送未做速率控制，網(wǎng)絡(luò)擁塞；
- 解決方案：在客戶端發(fā)送每個數(shù)據(jù)包后添加
time.sleep(0.001)，控制發(fā)送速率。
3. 數(shù)據(jù)包解析錯誤
- 原因：分隔符（|）與文件內(nèi)容沖突；
- 解決方案：改用更復雜的分隔符（如
|||或隨機字符串），或?qū)︻^部做Base64編碼。
4. 端口被占用
- 解決方案：修改
UDP_PORT（如改為10000），并在服務(wù)端添加SO_REUSEADDR選項（代碼已包含）。
到此這篇關(guān)于Python實現(xiàn)基于UDP的文件傳輸?shù)娜^程的文章就介紹到這了,更多相關(guān)Python基于UDP的文件傳輸內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
在本篇文章里小編給大家整理的是關(guān)于Python變量作用域的相關(guān)知識點內(nèi)容，需要的朋友們可以學習下。2020-06-06
Python內(nèi)置函數(shù)range()是一個用于生成一系列連續(xù)的整數(shù)的函數(shù)，它常用于循環(huán)結(jié)構(gòu)中，用于指定循環(huán)的次數(shù)或迭代的范圍，這篇文章主要介紹了python之內(nèi)置函數(shù)range,需要的朋友可以參考下2023-07-07
Python的經(jīng)緯度與xy坐標系相互轉(zhuǎn)換方式
這篇文章主要介紹了Python的經(jīng)緯度與xy坐標系相互轉(zhuǎn)換方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-02-02
本人對于壁紙一直偏佛系，不愛特意去找一堆壁紙。因此用Python簡單地搞了一個自動下載壁紙，定時隨機輪換的功能來自娛自樂，順便分享給大家。2020-10-10
利用python實現(xiàn).dcm格式圖像轉(zhuǎn)為.jpg格式
今天小編就為大家分享一篇利用python實現(xiàn).dcm格式圖像轉(zhuǎn)為.jpg格式，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-01-01
python opencv實現(xiàn)圖片旋轉(zhuǎn)矩形分割
這篇文章主要為大家詳細介紹了python opencv實現(xiàn)圖片旋轉(zhuǎn)矩形分割，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-07-07
Python基于Gensim實現(xiàn)文本相似度/匹配/查重
Gensim是基于Python語言的自然語言處理庫,用來主題建模、文本相似度等文本處理任務(wù),下面我們就來看看如何使用Gensim實現(xiàn)文本相似度/匹配/查重等操作吧2024-03-03
Python中涉及多線程的主要模塊包括threading、thread和concurrent.futures,現(xiàn)代Python編程推薦使用threading和concurrent.futures,以提供更高層次的抽象和可用性,感興趣的朋友跟隨小編一起看看吧2024-09-09
用Python爬取618當天某東熱門商品銷量數(shù)據(jù),看看大家喜歡什么!
618購物節(jié),準備分析一波購物節(jié)大家都喜歡買什么？本文以某東為例,Python爬取618活動的暢銷商品數(shù)據(jù),并進行數(shù)據(jù)清洗,最后以可視化的方式從不同角度去了解暢銷商品中,名列前茅的商品是哪些？銷售數(shù)據(jù)如何？用戶好評如何？等等,需要的朋友可以參考下2021-06-06
形態(tài)學處理方法是基于對二進制圖像進行處理的，卷積核決定圖像處理后的效果。本文將為大家詳細介紹一下OpenCV中的圖像形態(tài)學，感興趣的可以了解一下2022-01-01