背景

最近在開發(fā)工作中遇到需要兩臺(tái)本地設(shè)備之間進(jìn)行視頻流的傳輸?shù)那闆r。但是團(tuán)隊(duì)一來沒有這方面的專業(yè)人才，二來視頻流的傳續(xù)數(shù)據(jù)量很大，針對(duì)TCP和UDP的具體選擇也不明確。

本文是在上訴背景之下進(jìn)行的研究和開發(fā)工作。

目錄

背景

UDP和TCP協(xié)議的選擇

Socket-UDP協(xié)議代碼詳解

UDP協(xié)議發(fā)送端

UDP協(xié)議接收端

---

UDP和TCP協(xié)議的選擇

視頻流更加適合UDP協(xié)議的傳輸！

序號(hào)	UDP協(xié)議	TCP協(xié)議
是否需要握手	不需要握手	需要握手
是否確保數(shù)據(jù)幀傳輸準(zhǔn)確性	不確保	確保
是否確保數(shù)據(jù)幀傳輸順序一致性	不確保	確保
速度	相對(duì)高速	相對(duì)低速

對(duì)于一些需要準(zhǔn)確傳輸?shù)男畔?#xff0c;則選擇TCP協(xié)議

對(duì)于一些需要高速傳輸和不在意準(zhǔn)確性的數(shù)據(jù)，選擇UDP協(xié)議。

顯然實(shí)時(shí)視頻流就是一個(gè)典型的適合UDP協(xié)議的數(shù)據(jù)。

• 實(shí)時(shí)視頻流不在意數(shù)據(jù)是否完整傳輸（因?yàn)閭鬏斿e(cuò)誤的幀馬上就是過去式了，接著顯示新的幀）
• 實(shí)時(shí)視頻流不在意幀是否順序一致（少數(shù)的幾幀在短暫的時(shí)間戳內(nèi)順序不一致無傷大雅）
• 但實(shí)時(shí)視頻流需要幀高速

Socket-UDP協(xié)議代碼詳解

UDP協(xié)議發(fā)送端

是否常常遇到問題：

OSError: [WinError 10040] 一個(gè)在數(shù)據(jù)報(bào)套接字上發(fā)送的消息大于內(nèi)部消息緩沖區(qū)或其他一些網(wǎng)絡(luò)限制，或該用戶用于接收數(shù)據(jù)報(bào)的緩沖區(qū)比數(shù)據(jù)報(bào)小？

簡(jiǎn)單，發(fā)送端僅需要分片發(fā)送即可，當(dāng)然接收端也需要想要調(diào)整。

import cv2
import socket
import time
import struct
import numpy as np
import logging# 配置攝像頭和UDP傳輸參數(shù)
FPS_INTERVAL = 0.1  # 每隔0.1秒計(jì)算一次幀率
UDP_IP = "127.0.0.1"  # 目標(biāo)接收端IP
UDP_PORT = 12345  # 目標(biāo)接收端端口
MAX_UDP_SIZE = 1024  # 每個(gè)數(shù)據(jù)包最大傳輸大小，調(diào)整為1024字節(jié)# 設(shè)置日志
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger()# 創(chuàng)建UDP socket
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)# 打開攝像頭
cap = cv2.VideoCapture(1)
if not cap.isOpened():logger.error("無法打開攝像頭，請(qǐng)檢查設(shè)備連接")exit(1)cap.set(cv2.CAP_PROP_FRAME_WIDTH, 4000)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 3000)
cap.set(cv2.CAP_PROP_FPS, 60)# 初始化一些變量
frame_count = 0
last_time = time.time()while True:try:ret, frame = cap.read()if not ret:logger.warning("無法讀取攝像頭幀")continue  # 如果讀取失敗，跳過本次循環(huán)# 定義新的大小（寬度，高度）new_dims = (1280, 960)  # 新的寬度和高度# 使用cv2.resize()調(diào)整圖像大小frame = cv2.resize(frame, new_dims, interpolation=cv2.INTER_LINEAR)# 將幀轉(zhuǎn)換為JPEG格式ret, jpeg = cv2.imencode('.jpg', frame)if ret:# 將JPEG圖像數(shù)據(jù)轉(zhuǎn)為字節(jié)流data = jpeg.tobytes()data_len = len(data)# 發(fā)送幀數(shù)據(jù)的總長(zhǎng)度try:sock.sendto(struct.pack("L", data_len), (UDP_IP, UDP_PORT))  # 發(fā)送數(shù)據(jù)長(zhǎng)度except socket.error as e:logger.error(f"發(fā)送數(shù)據(jù)長(zhǎng)度失敗: {e}")continue  # 如果發(fā)送失敗，跳過本次循環(huán)# 分片發(fā)送數(shù)據(jù)for i in range(0, data_len, MAX_UDP_SIZE):packet = data[i:i+MAX_UDP_SIZE]try:sock.sendto(packet, (UDP_IP, UDP_PORT))  # 發(fā)送數(shù)據(jù)片段except socket.error as e:logger.error(f"發(fā)送數(shù)據(jù)片段失敗: {e}")continue  # 如果發(fā)送失敗，跳過本次循環(huán)# 計(jì)算幀率：每幀計(jì)算一次current_time = time.time()frame_time = current_time - last_time  # 計(jì)算當(dāng)前幀的時(shí)間差fps = 1.0 / frame_time if frame_time > 0 else 0  # 幀率 = 1 / 幀間隔# 更新上次幀的時(shí)間last_time = current_time# 在左上角顯示幀率cv2.putText(frame, f"client-FPS: {fps:.2f}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)# 顯示視頻流cv2.imshow('Camera', frame)except (cv2.error, socket.error) as e:logger.error(f"發(fā)生異常: {e}")# 如果發(fā)生異常，等待一段時(shí)間重試time.sleep(2)continue# 按'q'退出if cv2.waitKey(1) & 0xFF == ord('q'):break# 釋放資源
cap.release()
cv2.destroyAllWindows()

UDP協(xié)議接收端

在遇到緩存問題的時(shí)候，接收端除了增設(shè)分片接受以外，還需要進(jìn)行緩沖區(qū)大小的設(shè)定，這里推薦為5MB。當(dāng)然還需要try except之后清空所有的緩沖區(qū)！

import cv2
import socket
import struct
import numpy as np
import time
import logging# 配置UDP接收參數(shù)
UDP_IP = "127.0.0.1"  # 本地IP
UDP_PORT = 12345  # 端口號(hào)# 創(chuàng)建UDP socket
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((UDP_IP, UDP_PORT))# 增加接收緩沖區(qū)的大小
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1048576 * 5)  # 設(shè)置接收緩沖區(qū)大小為5MB# 用于接收數(shù)據(jù)的緩沖區(qū)
buffer = b''# 設(shè)置日志記錄
LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s"
logging.basicConfig(filename='udp_server.log', level=logging.DEBUG, format=LOG_FORMAT)last_time = time.time()def clear_socket_buffer():"""清空socket的接收緩沖區(qū)，丟棄所有未處理的數(shù)據(jù)。"""while True:# 嘗試讀取一部分?jǐn)?shù)據(jù)sock.settimeout(0.1)  # 設(shè)置一個(gè)短暫的超時(shí)避免阻塞try:data = sock.recv(4096)  # 嘗試讀取最大4KB的數(shù)據(jù)if not data:breakexcept socket.timeout:break  # 如果超時(shí)，退出循環(huán)while True:try:# 接收數(shù)據(jù)長(zhǎng)度（最多接收4字節(jié)）data_len, addr = sock.recvfrom(4)if not data_len:continuedata_len = struct.unpack("L", data_len)[0]# 接收?qǐng)D像數(shù)據(jù)（分片）buffer = b''  # 清空之前的緩沖區(qū)while len(buffer) < data_len:packet, addr = sock.recvfrom(1450)  # 每次接收一個(gè)片段buffer += packet  # 將接收到的數(shù)據(jù)片段拼接到緩沖區(qū)# 確保接收到完整數(shù)據(jù)if len(buffer) == data_len:# 解碼圖像nparr = np.frombuffer(buffer, np.uint8)frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)if frame is not None:# 計(jì)算并顯示幀率fps = 1 / (time.time() - last_time) if (time.time() - last_time) > 0 else 0last_time = time.time()# 在左上角顯示幀率cv2.putText(frame, f"Server-FPS: {fps:.2f}", (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)# 顯示接收到的圖像cv2.imshow('Received Video Stream', frame)else:logging.warning("接收到的圖像無法解碼！")continue  # 如果解碼失敗，跳過本次循環(huán)else:logging.error(f"接收到的數(shù)據(jù)包大小不匹配: 期望 {data_len} 字節(jié), 實(shí)際 {len(buffer)} 字節(jié)")continue  # 如果數(shù)據(jù)不完整，跳過本次循環(huán)except socket.timeout:logging.warning("接收超時(shí)，等待下一幀數(shù)據(jù)...")continue  # 如果超時(shí)，繼續(xù)等待except Exception as e:clear_socket_buffer()logging.error(f"發(fā)生異常: {e}")time.sleep(1)  # 如果發(fā)生異常，休眠2秒后繼續(xù)嘗試# 按 'q' 鍵退出if cv2.waitKey(1) & 0xFF == ord('q'):break# 釋放資源
cv2.destroyAllWindows()
sock.close()
logging.info("服務(wù)端退出，釋放資源")

其實(shí)直接拿去用即可！?