Bài trước trong series dùng Scrapy, một framework đã có sẵn bộ lập lịch và giới hạn đồng thời. Nhưng nhiều tác vụ nhỏ hơn, như gọi vài trăm API nội bộ hay kiểm tra trạng thái nghìn URL, không đáng để dựng cả framework. Khi đó thư viện chuẩn asyncio kết hợp với httpx cho phép một tiến trình đơn luồng giữ hàng chục kết nối mở cùng lúc. Khó không nằm ở việc chạy nhanh mà ở chỗ chạy nhanh vừa đủ: không làm sập máy chủ đích, không bị chặn IP, không tràn bộ nhớ. Bài này đi từ vòng lặp sự kiện tới một crawler hoàn chỉnh, mọi số liệu đều đo thật trên máy.
Bài nối tiếp Scrapy từ A tới Z: spider, item pipeline và middleware, Crawl data bằng Python: requests, BeautifulSoup và robots.txt và Môi trường Python cho dự án dữ liệu và AI.
Môi trường và cách đo
Toàn bộ code nằm trong thư mục code/s7-04, chạy trên Apple M4, 16 GB RAM, Python 3.10 (dòng phát hành tháng 10/2021, ổn định mới nhất thời điểm bài đăng; máy chạy dùng bản vá mới nhất của dòng 3.10, còn các gói được ghim ở phiên bản đầu năm 2022) với httpx==0.22.0 và beautifulsoup4==4.11.1 (xem requirements.txt). Để so sánh công bằng, phần lớn thí nghiệm chạy với một máy chủ HTTP cục bộ do ta tự viết (local_server.py): nó có endpoint trễ có chủ đích và đếm số request đang xử lý đồng thời ở phía server. Đo ở phía server quan trọng hơn đo ở client, vì mục tiêu của giới hạn đồng thời là bảo vệ phía bên kia. Phần chạy trên Internet chỉ dùng books.toscrape.com.
Vì sao asyncio phù hợp với crawl
Một request HTTP tốn phần lớn thời gian để chờ: chờ bắt tay TCP/TLS, chờ máy chủ xử lý, chờ dữ liệu về. CPU gần như rảnh. Đây là bài toán I/O-bound, đúng sở trường của asyncio. Với requests, mỗi lời gọi chặn cả luồng, nên 60 request mỗi cái 200 ms mất ít nhất 12 giây. Có hai hướng để chồng các khoảng chờ lên nhau: nhiều luồng (ThreadPoolExecutor) hoặc một vòng lặp sự kiện như asyncio. Theo tài liệu asyncio của Python, thư viện này dùng cú pháp async/await để viết code đồng thời, và rất hợp với mã mạng I/O-bound. Mỗi coroutine tự nhường quyền tại điểm await, vòng lặp chuyển sang coroutine khác đang sẵn sàng, không cần khoá giữa các luồng.
Cái giá là mọi thứ trên đường đi phải bất đồng bộ. Một lời gọi time.sleep(), một requests.get() hay một phép parse HTML nặng bên trong coroutine sẽ chặn cả vòng lặp, tức là chặn mọi tác vụ khác. Vì thế ta dùng httpx.AsyncClient thay cho requests, và asyncio.sleep() thay cho time.sleep().
Bước 1: chạy tuần tự, rồi gather không giới hạn
Hàm asyncio.gather() nhận nhiều awaitable, chạy đồng thời và trả kết quả theo đúng thứ tự đầu vào. Cách viết ngây thơ nhất là gom hết URL vào một lần gather:
async def fetch(client, url):
r = await client.get(url)
return r.status_code
async def unbounded(client, urls):
return await asyncio.gather(*(fetch(client, u) for u in urls))Cách này nhanh đến mức nguy hiểm. Ta đo với 60 request, mỗi request server giữ 200 ms, httpx.Limits được nâng rất cao để chỉ còn code của ta quyết định mức đồng thời. Kết quả chạy thật (demo1_semaphore.py):
60 request, mỗi request server giữ 200 ms
tuần tự 12.40s ok=60/60 peak_server_inflight=1
gather không giới hạn 0.25s ok=60/60 peak_server_inflight=60
Semaphore(5) 2.48s ok=60/60 peak_server_inflight=5
Semaphore(10) 1.29s ok=60/60 peak_server_inflight=10
Semaphore(20) 0.69s ok=60/60 peak_server_inflight=20Cột cuối là con số quan trọng: gather không giới hạn làm server thấy 60 request cùng lúc, và nếu danh sách là 60.000 URL thì thành chừng ấy kết nối đồng thời, đủ để cạn file descriptor ở client hoặc bị phía kia coi là tấn công.
Bước 2: giới hạn đồng thời bằng asyncio.Semaphore
Công cụ chuẩn để giới hạn là asyncio.Semaphore. Theo tài liệu, semaphore quản lý một bộ đếm nội bộ: mỗi acquire() giảm một, mỗi release() tăng một, và khi bộ đếm bằng 0 thì acquire() chờ. Dạng khuyến nghị là async with sem:, tương đương acquire và release trong try/finally:
async def bounded(client, urls, limit):
sem = asyncio.Semaphore(limit)
async def one(u):
async with sem:
return await fetch(client, u)
return await asyncio.gather(*(one(u) for u in urls))Bảng trên cho thấy hiệu quả rõ ràng: Semaphore(5) cho đúng 5 request đồng thời ở server, thời gian gần bằng lý thuyết 60 ÷ 5 × 0,2 = 2,4 giây (đo được 2,48). Tăng gấp đôi giới hạn thì thời gian giảm một nửa, đúng quy luật khi nút thắt là độ trễ. Hai chi tiết dễ sai:
- Tạo semaphore bên trong coroutine hoặc sau khi vòng lặp chạy. Ở Python 3.10 tham số
loopđã bị gỡ, và semaphore gắn với vòng lặp hiện hành khi dùng lần đầu. Tạo ở phạm vi module rồi dùng vớiasyncio.run()nhiều lần dễ gây lỗi "attached to a different loop". - Semaphore của asyncio không phải thread-safe, tài liệu ghi rõ điều này.
Để phát hiện lỗi release() thừa, dùng BoundedSemaphore: nó ném ValueError khi bộ đếm vượt giá trị khởi tạo.
Hai tầng giới hạn: asyncio.Semaphore và httpx.Limits
Bản thân httpx.AsyncClient cũng có giới hạn riêng qua httpx.Limits. Theo tài liệu giới hạn tài nguyên của httpx (và đã kiểm tra trong mã nguồn bản 0.22.0 dùng ở bài), mặc định max_connections=100, max_keepalive_connections=20 và keepalive_expiry=5 giây. Nghĩa là dù ta tạo 10.000 coroutine, client chỉ mở tối đa 100 kết nối, các request còn lại xếp hàng chờ trong pool. Nghe như đã giới hạn sẵn, nhưng có ba khác biệt:
- Limits của client tính cho mọi host, còn thường ta cần giới hạn theo từng host. Crawl 5 site cùng lúc mà dùng chung 100 kết nối, một site chậm có thể chiếm hết pool của site khác.
- Request đang chờ trong pool vẫn bị tính vào timeout pool. Chờ quá lâu sẽ gặp
PoolTimeoutthay vì được xếp hàng vô hạn. - Semaphore ở tầng ứng dụng cho phép gắn thêm logic: giãn cách, ưu tiên, thống kê.
Thí nghiệm sau cho thấy cái bẫy thứ hai. Pool chỉ có 2 kết nối, 6 request đồng thời, mỗi request server giữ 600 ms, thời gian chờ pool đặt 0,3 giây (demo2_errors.py, phần B):
--- B. PoolTimeout: pool chỉ có 2 kết nối, 6 request đồng thời, chờ pool tối đa 0.3s
#0: 200
#1: 200
#2: PoolTimeout
#3: PoolTimeout
#4: PoolTimeout
#5: PoolTimeoutBốn request không hề bị lỗi mạng, chúng chỉ chờ lấy kết nối quá lâu. Bài học: để Semaphore là giới hạn chặt hơn, và đặt max_connections bằng hoặc lớn hơn giá trị semaphore. Khi đó request luôn có sẵn kết nối và timeout chỉ còn phản ánh tốc độ thực của server.
Bốn loại timeout của httpx
Không đặt timeout là sai lầm phổ biến nhất của crawler bất đồng bộ: một server treo kết nối sẽ giữ một slot của semaphore mãi mãi, và sau vài lần như vậy toàn bộ crawler đứng im mà không báo lỗi. Theo tài liệu httpx, có bốn loại timeout riêng: connect (chờ bắt tay kết nối), read (chờ một khối dữ liệu đến), write (chờ gửi dữ liệu đi) và pool (chờ lấy kết nối từ pool). Mặc định là 5 giây cho cả bốn loại (đã kiểm tra: DEFAULT_TIMEOUT_CONFIG = Timeout(timeout=5.0) trong httpx 0.22.0). Lưu ý: read là thời gian chờ giữa hai khối dữ liệu, không phải tổng thời gian tải, nên khi cần chặn tổng thời gian, bọc thêm asyncio.wait_for().
timeout = httpx.Timeout(1.0, pool=0.5) # 1 s cho connect/read/write, 0.5 s cho pool
async with httpx.AsyncClient(timeout=timeout) as client:
await client.get(f"{base}/hang") # server treo 3 giây--- A. Các loại timeout của httpx
ReadTimeout sau 1.01s (server treo 3s)Tất cả các lớp này kế thừa httpx.TimeoutException, nên một mệnh đề except httpx.TimeoutException bắt được cả bốn. Còn asyncio.wait_for(aw, timeout) thì khi hết hạn sẽ huỷ tác vụ con và ném asyncio.TimeoutError. Tài liệu lưu ý nó chờ việc huỷ hoàn tất nên tổng thời gian có thể lớn hơn timeout một chút. Ta kiểm chứng tác vụ con của asyncio thật sự nhận CancelledError:
--- G. wait_for huỷ task con khi quá hạn
TimeoutError, task con đã nhận CancelledError: TrueRetry có backoff, jitter và tôn trọng Retry-After
Lỗi tạm thời là chuyện thường: 503 khi server quá tải, 429 khi bị giới hạn tốc độ, đứt kết nối giữa chừng. Retry ngay lập tức là cách tệ nhất vì làm tải tăng đúng lúc server đang yếu. Bài viết Exponential Backoff and Jitter của AWS Architecture Blog giải thích: chỉ backoff luỹ thừa thì các client vẫn thử lại thành từng đợt đồng bộ, giải pháp là thêm jitter (ngẫu nhiên hoá thời gian chờ). Trong code, ta chờ base × 2^(lần-1) × random(0,5…1,0) giây. Riêng 429 và 503, nếu server gửi header Retry-After thì phải theo nó: RFC 7231 mục 7.1.3 quy định giá trị là một số giây nguyên hoặc một ngày giờ HTTP, cho biết bao lâu client nên chờ trước khi gửi lại.
Bản demo xử lý cả hai dạng giá trị (số giây và ngày giờ HTTP bằng email.utils.parsedate_to_datetime) và chặn trần 60 giây để một server trả Retry-After: 3600 không giữ slot semaphore cả giờ; chạy thử parse_retry_after cho kết quả "1" thành 1.0, "3600" thành 60.0, ngày giờ sau 5 giây thành khoảng 4,76 và chuỗi sai định dạng thành None. Hàm fetch_with_retry trong fetcher.py (phần import datetime, timezone, parsedate_to_datetime lược bớt):
RETRY_STATUS = {429, 500, 502, 503, 504}
MAX_RETRY_AFTER = 60.0 # trần: server đòi chờ 3600 giây cũng không giữ slot cả giờ
def parse_retry_after(value):
"""Số giây nguyên hoặc ngày giờ HTTP (RFC 7231, mục 7.1.3) -> số giây, có trần; sai định dạng -> None."""
if not value:
return None
value = value.strip()
if value.isdigit():
secs = float(value)
else:
try:
when = parsedate_to_datetime(value)
except (TypeError, ValueError):
return None
if when.tzinfo is None:
when = when.replace(tzinfo=timezone.utc)
secs = max(0.0, (when - datetime.now(timezone.utc)).total_seconds())
return min(secs, MAX_RETRY_AFTER)
async def fetch_with_retry(client, url, *, retries=3, base_delay=0.2, log=None):
for attempt in range(1, retries + 2):
try:
r = await client.get(url)
except (httpx.TimeoutException, httpx.TransportError) as exc:
reason, wait = type(exc).__name__, None
else:
if r.status_code not in RETRY_STATUS:
return r, attempt
reason = f"HTTP {r.status_code}"
wait = parse_retry_after(r.headers.get("Retry-After"))
if attempt > retries:
return None, attempt
delay = wait if wait is not None else base_delay * (2 ** (attempt - 1)) * random.uniform(0.5, 1.0)
await asyncio.sleep(delay)Kết quả chạy với endpoint /flaky (lỗi 503 hai lần đầu), /limited (429 kèm Retry-After: 1) và một đường dẫn 404:
--- C. Retry với backoff: /flaky lỗi 503 hai lần rồi thành công
lần 1 lỗi (HTTP 503), chờ 0.13s
lần 2 lỗi (HTTP 503), chờ 0.23s
kết quả: 200 'ok sau 3 lan' sau 3 lần thử
--- D. Tôn trọng Retry-After: 429 kèm 'Retry-After: 1'
lần 1 lỗi (HTTP 429), chờ 1.00s
kết quả: 200 sau 2 lần thử, tổng 1.00s
--- E. 404 không retry
kết quả: 404 sau 1 lần thửThời gian chờ tăng gần gấp đôi và lệch ngẫu nhiên (0,13 rồi 0,23 giây); với 429 ta chờ đúng 1,00 giây theo yêu cầu của server; 404 là lỗi vĩnh viễn nên không retry. Đặt retry bên trong async with như crawler dưới đây thì lúc chờ backoff slot vẫn bị giữ, nên tổng tải giảm khi server yếu.
asyncio.gather, return_exceptions và cái bẫy lỗi im lặng
Hành vi mặc định của gather thường gây bất ngờ. Với return_exceptions=False, ngoại lệ đầu tiên được ném ngay cho nơi gọi, nhưng các tác vụ còn lại không bị huỷ mà vẫn chạy tiếp ngầm. Tài liệu viết rõ điều này. Với return_exceptions=True, ngoại lệ được coi như kết quả bình thường và nằm trong danh sách trả về. Với crawler gom hàng nghìn URL, gần như luôn nên dùng True để một URL hỏng không làm mất kết quả của các URL còn lại:
--- F. gather mặc định vs return_exceptions=True
mặc định: ngoại lệ nổi lên ngay -> lỗi giả lập
return_exceptions=True -> ['1', "RuntimeError('lỗi giả lập')", '2']Nếu chính lời gọi gather bị huỷ (ví dụ người dùng nhấn Ctrl+C), mọi tác vụ con chưa xong đều bị huỷ theo. Đây là cơ chế ta dựa vào để dừng gọn.
Hàng đợi và worker trong asyncio: giới hạn cả bộ nhớ
Có một điểm yếu ẩn trong cách gather(*(one(u) for u in urls)): nó tạo tất cả coroutine ngay từ đầu. Với 1 triệu URL, bộ nhớ cho 1 triệu đối tượng task là đáng kể dù chỉ có 10 cái chạy thật. Mẫu hàng đợi + N worker giải quyết cả hai vấn đề: chỉ có N tác vụ tồn tại, và đồng thời tối đa là N. Dưới đây là crawler.py, gồm một bộ giới hạn theo host (semaphore cộng giãn cách tối thiểu giữa hai lần bắt đầu request) và vòng đời worker đúng cách:
class HostLimiter:
def __init__(self, concurrency, min_interval):
self.sem = asyncio.Semaphore(concurrency)
self.min_interval = min_interval
self._lock = asyncio.Lock()
self._next = 0.0
async def __aenter__(self):
await self.sem.acquire()
async with self._lock:
now = time.monotonic()
wait = max(0.0, self._next - now)
self._next = max(now, self._next) + self.min_interval
if wait:
await asyncio.sleep(wait)
async def __aexit__(self, *exc):
self.sem.release()
async def worker(queue, client, limiter, results, stats):
while True:
url = await queue.get()
try:
async with limiter:
r, attempts = await fetch_with_retry(client, url)
... # parse bằng BeautifulSoup, ghi vào results
finally:
queue.task_done()Phần điều phối nằm ở hàm crawl. Worker sống vô hạn (while True), nên khi queue.join() báo hết việc ta phải huỷ chúng tường minh, rồi gather(..., return_exceptions=True) để chờ chúng dọn xong:
async with httpx.AsyncClient(headers=headers, timeout=httpx.Timeout(10.0),
follow_redirects=True,
limits=httpx.Limits(max_connections=concurrency)) as client:
tasks = [asyncio.create_task(worker(queue, client, limiter, results, stats))
for _ in range(workers)]
try:
await queue.join()
finally:
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)Chú ý ba điều. queue.task_done() nằm trong finally, nếu quên thì join() treo mãi khi một URL gây lỗi. Danh sách tasks giữ tham chiếu mạnh tới task, vì tài liệu nhấn mạnh vòng lặp chỉ giữ tham chiếu yếu nên một task không được tham chiếu có thể bị thu gom rác giữa chừng. Và một AsyncClient dùng chung cho mọi worker, đúng khuyến nghị của httpx: tạo client cho từng request làm mất lợi ích của connection pooling.
Chạy thật trên books.toscrape.com, 20 trang danh mục (mỗi trang 20 sách), giãn cách tối thiểu 0,1 giây giữa hai lần bắt đầu request, thay đổi số kết nối đồng thời (crawler.py):
concurrency=1 trang=20 lỗi=0 sách=400 thời gian=11.25s
ví dụ: A Light in the Attic £51.77
concurrency=4 trang=20 lỗi=0 sách=400 thời gian=3.38s
ví dụ: A Light in the Attic £51.77
concurrency=10 trang=20 lỗi=0 sách=400 thời gian=2.62s
ví dụ: In Her Wake £12.84Từ 1 lên 4 đồng thời, thời gian giảm hơn một nửa. Từ 4 lên 10, mức giảm nhỏ lại: cái trần tính được là 20 trang × 0,1 giây giãn cách = 2 giây, và các lần chạy đã gần chạm trần đó. Đây là điều cần nhớ: tăng đồng thời không tăng tốc mãi. Sau một ngưỡng, nút thắt chuyển sang giãn cách, băng thông hoặc chính server đích, và ta chỉ nhận thêm rủi ro bị chặn.
Crawl có trách nhiệm
Tốc độ của asyncio khiến một script nhỏ cũng đủ sức gây hại, nên các nguyên tắc sau là bắt buộc, không phải tuỳ chọn:
- Đọc robots.txt và điều khoản sử dụng trước khi crawl bất kỳ site nào, như đã nói ở bài về
requests. Ví dụ trong bài chỉ dùngbooks.toscrape.com(trang dành cho việc học, hiện không có robots.txt và trả 404 cho đường dẫn này) và máy chủ cục bộ. - Khai báo User-Agent rõ ràng kèm cách liên hệ, như trong
crawler.py, để quản trị viên đối tác có thể nhắn khi có vấn đề. - Giới hạn theo từng host, bắt đầu với 1–4 request đồng thời và giãn cách 0,5–1 giây với site không phải của mình; dừng khi liên tục nhận 429.
- Không thu thập dữ liệu cá nhân nếu không có cơ sở pháp lý rõ ràng.
Phần rate limit, cache và ETag sẽ được đào sâu ở bài cuối của nhóm crawl trong series.
Khi nào không nên dùng asyncio
- Vài chục URL, chạy một lần:
requeststuần tự đơn giản hơn nhiều và đủ nhanh, không đáng để chịu độ phức tạp của asyncio. - Phần lớn thời gian nằm ở xử lý CPU (parse HTML khổng lồ, regex nặng): asyncio không giúp gì, thậm chí làm vòng lặp bị chặn. Chuyển phần nặng sang
run_in_executorvớiProcessPoolExecutor. - Cần cả hệ sinh thái crawl (lập lịch, dedupe URL, pipeline, retry, thống kê, xuất dữ liệu): Scrapy, xây trên Twisted, đã giải sẵn các việc này như bài trước đã trình bày.
- Trang cần chạy JavaScript:
httpxchỉ lấy HTML thô. Bài kế tiếp xử lý trường hợp này bằng trình duyệt thật.
Lỗi thường gặp
| Triệu chứng | Nguyên nhân | Cách sửa |
|---|---|---|
| Crawler chạy tuần tự dù đã dùng async | Gọi requests.get hoặc time.sleep trong coroutine, hoặc await từng tác vụ trong vòng for | Dùng httpx.AsyncClient, asyncio.sleep; gom bằng gather hoặc worker |
| Treo vĩnh viễn, không báo lỗi | Không đặt timeout; task_done() không nằm trong finally | Luôn đặt httpx.Timeout; dùng try/finally |
PoolTimeout hàng loạt | Semaphore lớn hơn max_connections | Đặt semaphore nhỏ hơn hoặc bằng pool |
| Một URL lỗi làm mất cả lô kết quả | gather mặc định ném ngoại lệ đầu tiên | return_exceptions=True hoặc bắt lỗi trong worker |
| Bị chặn IP hoặc nhận nhiều 429 | Tốc độ quá cao, bỏ qua Retry-After | Giảm đồng thời, thêm giãn cách và backoff |
Checklist cho crawler asyncio
- Một
AsyncClientdùng chung, đóng bằngasync with; mọi lời gọi trong coroutine đều không chặn vòng lặp asyncio. - Timeout đủ bốn loại, và
wait_fornếu cần giới hạn tổng thời gian. - Giới hạn đồng thời theo host bằng semaphore;
max_connectionskhông nhỏ hơn giá trị đó. - Retry chỉ cho lỗi tạm thời (429, 5xx, lỗi mạng), có backoff, jitter, trần số lần và tôn trọng
Retry-After. - Dùng hàng đợi và số worker cố định khi số URL lớn.
- Huỷ và chờ worker dọn dẹp ở
finally. - User-Agent rõ ràng, tuân thủ robots.txt và điều khoản site.
Bài tiếp theo
Có những trang mà httpx chỉ nhận được khung HTML rỗng vì nội dung được dựng bằng JavaScript. Bài kế tiếp: Crawl trang render JavaScript bằng Playwright cho Python.
