Trong thư viện asyncio, asyncio.Event là cơ chế đồng bộ hóa luồng điều khiển đơn giản và trực quan nhất: nó hoạt động như một chiếc cờ hiệu (flag) hoặc một chiếc đèn giao thông nội bộ.
Nó cho phép một tác vụ (task) phát tín hiệu và nhiều tác vụ khác đứng đợi tín hiệu đó trước khi tiếp tục chạy.
asyncio.EventChiếc cờ hiệu này chỉ có 2 trạng thái: Bật (True) hoặc Tắt (False). Khi khởi tạo, mặc định cờ luôn ở trạng thái Tắt (False).
asyncio.Event xoay quanh 4 phương thức cốt lõi:
await event.wait(): Tác vụ gọi lệnh này sẽ tạm dừng và nhường CPU cho việc khác cho đến khi cờ được bật thành True. Nếu cờ đã là True từ trước, nó sẽ đi qua ngay mà không cần chờ.
event.set(): Bật cờ thành True. Ngay khi lệnh này chạy, tất cả các coroutine đang đợi tại wait() sẽ cùng lúc bừng tỉnh để chạy tiếp.
event.clear(): Hạ cờ về False (khóa cổng lại).
event.is_set(): Trả về True/False để kiểm tra trạng thái cờ hiện tại.
Hình ảnh ẩn dụ cho hai bạn nhỏ:
asyncio.Eventgiống hệt chiếc cờ lệnh của trọng tài ở vạch xuất phát cuộc đua. Các vận động viên (wait()) khởi động xong phải đứng xếp hàng chờ. Trọng tài phất cờ (set()), tất cả các vận động viên đồng loạt lao đi cùng một lúc!
Giả sử hệ thống cần:
1 tác vụ tải file cấu hình lớn từ mạng (mất 2 giây).
2 tác vụ công nhân (Worker A và Worker B) phải đợi tải xong cấu hình mới được phép xử lý dữ liệu.
import asyncio
import time
async def worker(ten, event_kich_hoat):
print(f"⏳ [{ten}]: Đã sẵn sàng, đang đứng đợi tín hiệu kích hoạt...")
# Tạm dừng tại đây, nhường CPU cho tác vụ khác
await event_kich_hoat.wait()
# Khi cờ được bật, dòng code dưới đây mới thực thi
print(f"🚀 [{ten}]: Nhận được cờ lệnh! Bắt đầu xử lý dữ liệu...")
await asyncio.sleep(1)
print(f"✅ [{ten}]: Đã hoàn thành công việc.")
async def he_thong_khoi_dong(event_kich_hoat):
print("📦 [HỆ THỐNG]: Đang kết nối máy chủ và tải cấu hình...")
await asyncio.sleep(2) # Giả lập tải dữ liệu mất 2 giây
print("🏁 [HỆ THỐNG]: Tải cấu hình thành công! Phất cờ bật đèn xanh...")
# Phất cờ: Đánh thức tất cả worker đang await event_kich_hoat.wait()
event_kich_hoat.set()
async def main():
# Khởi tạo cờ hiệu (mặc định trạng thái là False)
co_lenh = asyncio.Event()
# Gom các tác vụ chạy đồng thời
await asyncio.gather(
worker("Worker 1", co_lenh),
worker("Worker 2", co_lenh),
he_thong_khoi_dong(co_lenh)
)
if __name__ == "__main__":
asyncio.run(main())
----------------------
⏳ [Worker 1]: Đã sẵn sàng, đang đứng đợi tín hiệu kích hoạt...
⏳ [Worker 2]: Đã sẵn sàng, đang đứng đợi tín hiệu kích hoạt...
📦 [HỆ THỐNG]: Đang kết nối máy chủ và tải cấu hình...
(chờ 2 giây)
🏁 [HỆ THỐNG]: Tải cấu hình thành công! Phất cờ bật đèn xanh...
🚀 [Worker 1]: Nhận được cờ lệnh! Bắt đầu xử lý dữ liệu...
🚀 [Worker 2]: Nhận được cờ lệnh! Bắt đầu xử lý dữ liệu...
(chờ 1 giây)
✅ [Worker 1]: Đã hoàn thành công việc.
✅ [Worker 2]: Đã hoàn thành công việc.
Event, Lock và Queue| Công cụ | Mục đích chính | Cơ chế hoạt động |
asyncio.Event |
Phát tín hiệu (1 $\rightarrow$ Nhiều) | Một bên thông báo, tất cả các bên đang chờ đều được đi qua cùng lúc. Không truyền kèm dữ liệu. |
asyncio.Lock |
Loại trừ lẫn nhau (Mutual Exclusion) | Chỉ duy nhất 1 tác vụ được vào vùng tranh chấp (Critical Section) tại một thời điểm để tránh race condition. |
asyncio.Queue |
Truyền nhận dữ liệu (Producer $\rightarrow$ Consumer) | Đưa dữ liệu qua hàng đợi an toàn giữa các coroutine. |
---------------------------------------
threading.Event và asyncio.Event đều có chung giao diện lập trình (API) và mục đích sử dụng là quản lý một cờ hiệu nội bộ (Internal Flag) nhằm truyền tín hiệu giữa các đơn vị thực thi. Tuy nhiên, cơ chế kích hoạt và bản chất vật lý bên dưới của chúng hoàn toàn khác nhau.
| Tiêu chí | threading.Event | asyncio.Event |
| Môi trường hoạt động | Đa luồng thực sự (Multi-threading / OS Threads). | Đơn luồng bất đồng bộ (Single-thread / Coroutines). |
Hành vi khi gọi .wait() |
Chặn luồng hệ điều hành (Blocking): Thread bị chuyển sang trạng thái ngủ (Sleep/Waiting State) và rời khỏi hàng đợi CPU của OS. | Tạm dừng Coroutine không chặn (Non-blocking): Nhường CPU lại cho Event Loop; thread chính vẫn tiếp tục chạy các task khác. |
| Cơ chế đánh thức bên dưới | Dùng Condition Variable (threading.Condition) và Mutex Lock của tầng OS/C-runtime (pthread_cond_wait). |
Tạo ra một asyncio.Future rỗng và lưu vào danh sách chờ nội bộ của Python. |
Khi gọi .set() |
Kích hoạt tín hiệu notify_all() của OS để đánh thức tất cả các thread đang ngủ dậy tranh chấp CPU. |
Duyệt qua danh sách các Future đang chờ và gọi future.set_result(True) để đưa các Coroutine vào Ready Queue của Event Loop. |
| Tính an toàn đa luồng (Thread-Safety) | Thread-safe hoàn toàn: Có thể gọi .set() hoặc .wait() an toàn từ bất kỳ thread nào. |
Không Thread-safe: Chỉ an toàn khi gọi trong cùng một Thread chứa Event Loop đó. |
| Hỗ trợ Timeout | Hỗ trợ trực tiếp qua tham số: event.wait(timeout=2.0). |
Không có tham số timeout trong .wait(); phải bọc bằng asyncio.wait_for(event.wait(), timeout=2.0). |
threading.Event: Đánh thức qua biến điều kiện của OS (Condition Variable)Khi bạn gọi threading.Event().wait():
Thread hiện tại yêu cầu lấy một Lock nội bộ của biến điều kiện (Condition).
Nó đăng ký Thread ID của mình vào danh sách chờ của Kernel hệ điều hành.
Thread tự giải phóng khóa và đi vào trạng thái ngủ sâu (pthread_cond_wait trong C). CPU hoàn toàn bỏ qua thread này cho đến khi có một luồng khác gọi event.set().
Khi event.set() được gọi từ Thread B, hệ điều hành phát tín hiệu đánh thức tất cả các thread đang ngủ để tiếp tục chạy tiếp.
[Thread 1] ──> event.wait() ──> [OS Kernel: Đưa Thread 1 vào trạng thái Sleep]
▲
[Thread 2] ──> event.set() ──> [OS Kernel: Gửi tín hiệu đánh thức Thread 1]
asyncio.Event: Đánh thức qua danh sách Future trong Event LoopBên trong mã nguồn của asyncio.Event không hề có Mutex Lock hay System Call nào của OS. Nó chỉ là một cấu trúc dữ liệu thuần Python:
# Mô phỏng logic nội bộ của asyncio.Event
class AsyncioEventMock:
def __init__(self):
self._value = False
self._waiters = [] # Danh sách chứa các asyncio.Future
async def wait(self):
if self._value:
return True
fut = loop.create_future()
self._waiters.append(fut)
await fut # Coroutine nhường quyền tại đây
return True
def set(self):
self._value = True
for fut in self._waiters:
if not fut.done():
fut.set_result(True) # Đánh thức Coroutine thông qua Future
self._waiters.clear()Khi bạn gõ await event.wait():
Nó tạo một chiếc hộp rỗng Future và cất vào mảng self._waiters.
Coroutine của bạn dừng lại tại điểm await fut, nhường CPU cho Event Loop chạy các việc khác.
Khi ai đó gọi event.set(), mã nguồn chỉ đơn giản duyệt qua mảng _waiters và gọi fut.set_result(True).
Event Loop thấy các Future này đã xong liền đưa các Coroutine tương ứng trở lại hàng đợi sẵn sàng chạy.
Một lỗi kinh điển xảy ra khi một Thread phụ (ví dụ: Thread kéo RTSP Camera) muốn phát tín hiệu báo cho Event Loop chính (Asyncio Web Server) biết có frame mới bằng cách gọi trực tiếp asyncio_event.set():
# NGUY HIỂM: Gây hỏng trạng thái (Race Condition nội bộ) hoặc Crash
def camera_thread_worker(async_event):
# Đang chạy trong OS Thread riêng biệt
async_event.set() # SAI! asyncio.Event không thread-safeCách khắc phục chuẩn: Vì asyncio.Event không an toàn khi gọi xuyên luồng, bạn phải dùng loop.call_soon_threadsafe() để ủy thác việc giương cờ cho chính Event Loop thực thi:
def camera_thread_worker(loop, async_event):
# Đẩy lệnh set() an toàn xuyên luồng vào Event Loop
loop.call_soon_threadsafe(async_event.set)