Dưới đây là ví dụ code hoàn chỉnh kết hợp giữa camera_thread (chạy Threading) phát cờ hiệu báo cho stream_handler (chạy Asyncio) truyền dữ liệu qua WebSocket.
Dưới đây là một ví dụ hoàn chỉnh kết hợp giữa:
camera_thread (Chạy trên OS Thread): Đọc video liên tục bằng OpenCV, lưu frame mới nhất vào bộ đệm và dùng loop.call_soon_threadsafe(new_frame_event.set) để phát cờ hiệu.
stream_handler (Coroutine chạy trên asyncio): Lắng nghe cờ hiệu (await new_frame_event.wait()), mã hóa frame sang JPEG rồi phát qua WebSocket tới trình duyệt web theo thời gian thực.
Web Client (HTML/JS): Kết nối WebSocket và hiển thị luồng video trực tiếp trên thẻ <img>.
Cài đặt thư viện websockets và opencv-python:
pip install opencv-python websockets
camera_ws_server.pyTạo file camera_ws_server.py và chạy trên máy tính của bạn:
import asyncio
import cv2
import threading
import time
import websockets
# =====================================================================
# TẦNG DỮ LIỆU DÙNG CHUNG (SHARED BUFFER & SYNCHRONIZATION)
# =====================================================================
latest_jpeg_bytes = None
frame_lock = threading.Lock()
stop_thread_flag = False
# =====================================================================
# 1. CAMERA THREAD (CHẠY TRÊN OS THREAD ĐỘC LẬP)
# =====================================================================
def camera_worker(loop: asyncio.AbstractEventLoop, new_frame_event: asyncio.Event, source=0):
"""
Luồng phần cứng liên tục kéo frame từ Camera / Video.
Khi có frame mới, mã hóa JPEG và kích hoạt cờ hiệu sang Event Loop.
"""
global latest_jpeg_bytes, stop_thread_flag
cap = cv2.VideoCapture(source)
if not cap.isOpened():
print(f"❌ Không thể mở nguồn camera: {source}")
return
print("📷 [Camera Thread]: Đã kết nối Camera, bắt đầu thu hình...")
frame_idx = 0
try:
while not stop_thread_flag:
ret, frame = cap.read()
if not ret:
time.sleep(0.01)
continue
frame_idx += 1
# Vẽ timestamp và số thứ tự frame lên ảnh để kiểm chứng độ trễ
h, w = frame.shape[:2]
ts = time.strftime('%X')
cv2.putText(frame, f"LIVE | Frame: {frame_idx} | {ts}", (20, 40),
cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2)
# Nén ảnh thành định dạng JPEG
success, encoded_image = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 75])
if success:
# Ghi frame vào bộ đệm chia sẻ an toàn luồng
with frame_lock:
latest_jpeg_bytes = encoded_image.tobytes()
# PHÁT CỜ HIỆU THREAD-SAFE TỚI ASYNCIO EVENT LOOP
loop.call_soon_threadsafe(new_frame_event.set)
# Giữ nhịp đọc khoảng 30 FPS (~33ms/frame)
time.sleep(0.03)
finally:
cap.release()
print("📷 [Camera Thread]: Đã giải phóng tài nguyên Camera.")
# =====================================================================
# 2. STREAM HANDLER (COROUTINE CHẠY TRÊN ASYNCIO EVENT LOOP)
# =====================================================================
async def stream_handler(websocket, new_frame_event: asyncio.Event):
"""
Mỗi khi có Client kết nối WebSocket, hàm này sẽ chạy để truyền luồng frame.
Lắng nghe new_frame_event.wait() để đẩy frame đi ngay khi có ảnh mới.
"""
client_ip = websocket.remote_address
print(f"🌐 [WebSocket Server]: Client kết nối thành công: {client_ip}")
try:
while True:
# Coroutine tạm dừng tại đây nhường CPU cho đến khi Camera Thread giương cờ
await new_frame_event.wait()
# Hạ cờ xuống để đón nhận chu kỳ frame tiếp theo
new_frame_event.clear()
# Lấy bản sao JPEG bytes từ bộ đệm
frame_data = None
with frame_lock:
frame_data = latest_jpeg_bytes
if frame_data:
# Gửi trực tiếp chuỗi byte nhị phân (Binary) qua WebSocket
await websocket.send(frame_data)
except websockets.exceptions.ConnectionClosed:
print(f"🌐 [WebSocket Server]: Client ngắt kết nối: {client_ip}")
# =====================================================================
# 3. ĐIỀU PHỐI CHƯƠNG TRÌNH CHÍNH (MAIN ENTRYPOINT)
# =====================================================================
async def main():
loop = asyncio.get_running_loop()
new_frame_event = asyncio.Event()
# Khởi chạy Camera trên một OS Thread riêng biệt
# Thay 0 bằng đường dẫn file MP4 nếu máy không có Webcam
cam_thread = threading.Thread(
target=camera_worker,
args=(loop, new_frame_event, 0),
daemon=True
)
cam_thread.start()
# Tạo WebSocket Server lắng nghe tại cổng 8765
server_coro = websockets.serve(
lambda ws: stream_handler(ws, new_frame_event),
host="localhost",
port=8765
)
async with server_coro:
print("🚀 [WebSocket Server] Đang lắng nghe tại: ws://localhost:8765")
print("💡 Hãy mở file index.html trên trình duyệt để xem video.")
# Duy trì server chạy vô tận
await asyncio.Future()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
stop_thread_flag = True
print("\n🛑 Nhận tín hiệu dừng chương trình (Ctrl+C). Đang tắt...")index.htmlTạo một file index.html cùng thư mục (chỉ cần nhấp đúp để mở trực tiếp trên Chrome/Firefox):
<!DOCTYPE html>
<html lang="vi">
<head>
<meta charset="UTF-8">
<title>Live Video WebSocket Stream</title>
<style>
body {
font-family: Arial, sans-serif;
background-color: #1e1e1e;
color: #ffffff;
text-align: center;
padding: 30px;
}
#streamBox {
margin: 20px auto;
max-width: 800px;
border: 3px solid #00ff88;
border-radius: 8px;
overflow: hidden;
box-shadow: 0 4px 15px rgba(0, 0, 0, 0.5);
}
img {
width: 100%;
height: auto;
display: block;
}
.status {
font-size: 1.1rem;
color: #00ff88;
margin-bottom: 10px;
}
</style>
</head>
<body>
<h2>Hệ Thống Stream Camera: Threading + Asyncio WebSocket</h2>
<div class="status" id="statusText">Trạng thái: Đang kết nối...</div>
<div id="streamBox">
<img id="videoFeed" src="" alt="Đang tải luồng camera..." />
</div>
<script>
const statusText = document.getElementById("statusText");
const videoFeed = document.getElementById("videoFeed");
// Kết nối tới WebSocket server Python
const ws = new WebSocket("ws://localhost:8765");
ws.binaryType = "blob"; // Nhận dữ liệu dưới dạng Binary Blob
ws.onopen = () => {
statusText.innerText = "Trạng thái: Đã kết nối thời gian thực!";
statusText.style.color = "#00ff88";
};
ws.onmessage = (event) => {
// Chuyển blob nhị phân JPEG nhận được thành Object URL để hiển thị trên thẻ <img>
const blob = event.data;
const url = URL.createObjectURL(blob);
// Giải phóng URL cũ để tránh rò rỉ bộ nhớ trình duyệt
if (videoFeed.src.startsWith("blob:")) {
URL.revokeObjectURL(videoFeed.src);
}
videoFeed.src = url;
};
ws.onclose = () => {
statusText.innerText = "Trạng thái: Mất kết nối tới Server.";
statusText.style.color = "#ff4444";
};
ws.onerror = (err) => {
console.error("Lỗi WebSocket:", err);
statusText.innerText = "Trạng thái: Lỗi kết nối WebSocket.";
statusText.style.color = "#ff4444";
};
</script>
</body>
</html>Mở Terminal và chạy server:
python camera_ws_server.py
Mở file index.html bằng trình duyệt web.
Luồng hình ảnh từ webcam sẽ xuất hiện ngay lập tức trên trình duyệt với độ trễ (latency) gần như bằng không, không hề có hiện tượng giật hình hay nghẽn bộ đệm.
---------------------------------------------------------------------------------------
Hướng dẫn kỹ thuật phân phối (Broadcast) một luồng frame từ Camera Thread tới danh sách nhiều WebSocket clients kết nối đồng thời.
Để phát sóng (broadcast) một luồng camera tới nhiều WebSocket client cùng lúc, kiến trúc cần giải quyết 2 bài toán then chốt:
Quản lý danh sách kết nối: Một tập hợp (set) lưu giữ toàn bộ socket client đang hoạt động. Khi client mở kết nối thì thêm vào set, khi ngắt kết nối thì loại bỏ an toàn.
Tránh nghẽn kênh (Slow Client Problem): Không gửi dữ liệu tuần tự cho từng client (vì nếu 1 client mạng chậm sẽ kéo gián đoạn tất cả các client khác). Thay vào đó, dùng asyncio.gather() kết hợp cờ return_exceptions=True để gửi song song tới tất cả client cùng một thời điểm.
[Camera Thread (OS Thread)]
│ (1 frame / 33ms)
▼
[latest_jpeg_bytes (RAM Buffer)] ──> call_soon_threadsafe(new_frame_event.set)
│
▼
[Broadcaster Coroutine]
│
┌───────────────────────────┼───────────────────────────┐
▼ ▼ ▼
[Client 1 (Chrome)] [Client 2 (Firefox)] [Client 3 (Mobile)]
camera_broadcast_server.pyimport asyncio
import cv2
import threading
import time
import websockets
# =====================================================================
# 1. TẦNG DỮ LIỆU CHIA SẺ & DANH SÁCH CLIENTS
# =====================================================================
latest_jpeg_bytes = None
frame_lock = threading.Lock()
stop_thread_flag = False
# Tập hợp lưu trữ tất cả client WebSocket đang kết nối
# Cần quản lý trong cùng một Event Loop để đảm bảo thread-safe nội bộ của asyncio
connected_clients = set()
# =====================================================================
# 2. CAMERA THREAD (CHẠY TRÊN OS THREAD RIÊNG BIỆT)
# =====================================================================
def camera_worker(loop: asyncio.AbstractEventLoop, new_frame_event: asyncio.Event, source=0):
global latest_jpeg_bytes, stop_thread_flag
cap = cv2.VideoCapture(source)
if not cap.isOpened():
print(f"❌ Không thể mở nguồn camera: {source}")
return
print("📷 [Camera Thread]: Đã kết nối camera, bắt đầu thu hình...")
frame_idx = 0
try:
while not stop_thread_flag:
ret, frame = cap.read()
if not ret:
time.sleep(0.01)
continue
frame_idx += 1
# Vẽ thông tin kiểm chứng lên frame
ts = time.strftime('%X')
cv2.putText(frame, f"BROADCAST | Frame: {frame_idx} | {ts}", (20, 40),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2)
# Nén frame sang định dạng JPEG
success, encoded_img = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 70])
if success:
with frame_lock:
latest_jpeg_bytes = encoded_img.tobytes()
# Đánh thức Event Loop xử lý broadcast
loop.call_soon_threadsafe(new_frame_event.set)
# Giữ nhịp đọc khoảng 30 FPS (~33ms)
time.sleep(0.03)
finally:
cap.release()
print("📷 [Camera Thread]: Đã giải phóng camera.")
# =====================================================================
# 3. COROUTINE PHÁT SÓNG TRUNG TÂM (CENTRAL BROADCASTER)
# =====================================================================
async def frame_broadcaster(new_frame_event: asyncio.Event):
"""
Chỉ chạy duy nhất 1 coroutine này để lắng nghe cờ hiệu từ Camera Thread,
sau đó phát song song tới tất cả clients trong connected_clients.
"""
print("📡 [Broadcaster]: Bộ phát sóng trung tâm đã sẵn sàng.")
while True:
# Chờ frame mới từ Camera Thread
await new_frame_event.wait()
new_frame_event.clear()
# Không có client nào kết nối thì bỏ qua bước gửi
if not connected_clients:
continue
# Lấy dữ liệu frame an toàn từ RAM
with frame_lock:
frame_data = latest_jpeg_bytes
if not frame_data:
continue
# Gửi đồng thời tới toàn bộ client thông qua asyncio.gather
# return_exceptions=True giúp 1 client lỗi không làm sập các client khác
tasks = [client.send(frame_data) for client in connected_clients]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Xử lý các client bị đứt kết nối ngầm khi gửi
disconnected = set()
for client, res in zip(list(connected_clients), results):
if isinstance(res, Exception):
disconnected.add(client)
for client in disconnected:
connected_clients.discard(client)
print(f"⚠️ [Broadcaster]: Gỡ bỏ client lỗi kết nối: {client.remote_address}")
# =====================================================================
# 4. WEBSOCKET CONNECTION HANDLER
# =====================================================================
async def ws_handler(websocket):
"""Quản lý vòng đời kết nối của từng client."""
client_addr = websocket.remote_address
connected_clients.add(websocket)
print(f"🟢 [WebSocket]: Client mới kết nối: {client_addr} (Hiện có: {len(connected_clients)} viewers)")
try:
# Giữ kết nối mở và lắng nghe ping/pong hoặc tin nhắn đóng từ client
await websocket.wait_closed()
finally:
connected_clients.discard(websocket)
print(f"🔴 [WebSocket]: Client ngắt kết nối: {client_addr} (Còn lại: {len(connected_clients)} viewers)")
# =====================================================================
# 5. KHỞI CHẠY HỆ THỐNG
# =====================================================================
async def main():
loop = asyncio.get_running_loop()
new_frame_event = asyncio.Event()
# 1. Chạy Camera Thread
cam_thread = threading.Thread(
target=camera_worker,
args=(loop, new_frame_event, 0), # Thay 0 bằng file video nếu cần
daemon=True
)
cam_thread.start()
# 2. Khởi chạy Coroutine Broadcaster trung tâm ngầm
asyncio.create_task(frame_broadcaster(new_frame_event))
# 3. Khởi chạy WebSocket Server lắng nghe tại cổng 8765
async with websockets.serve(ws_handler, host="0.0.0.0", port=8765):
print("🚀 Server WebSocket Broadcast đang lắng nghe tại ws://localhost:8765")
print("💡 Bạn có thể mở đồng thời nhiều tab trình duyệt với file index.html để kiểm tra.")
await asyncio.Future() # Duy trì loop chạy mãi mãi
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
stop_thread_flag = True
print("\n🛑 Nhận tín hiệu dừng (Ctrl+C). Đang tắt server...")Chỉ 1 lần đọc và nén ảnh (Zero-Redundancy): Dù có 10 hay 100 client kết nối thì OpenCV trong Camera Thread chỉ nén ảnh JPEG đúng 1 lần duy nhất cho mỗi frame. Không bị nhân số lần xử lý CPU theo số lượng người xem.
Gửi phi tuần tự (True Concurrency): Dùng asyncio.gather(*tasks) giúp đẩy các khối byte JPEG vào buffer card mạng của tất cả các socket gần như cùng một tích tắc.
Tự phục hồi và dọn rác (Auto-Cleanup): Khi người dùng tắt tab trình duyệt, ws_handler tự động kích hoạt khối finally để xóa socket khỏi connected_clients, giải phóng triệt để RAM và kết nối socket.