Migrate từ ThreadPool sang AsyncIO trong Python

Bạn có một Python service đang dùng ThreadPoolExecutor để xử lý I/O concurrent. Nó chạy được — nhưng thread count cứ leo thang, memory ngày càng tăng, và thêm worker vào cũng chẳng cải thiện được gì. Bạn nghe nói asyncio "tốt hơn," nhưng codebase toàn executor.submit() và không biết bắt đầu từ đâu.

Bài này đi thẳng vào migration thực tế: khi nào nên làm, cách refactor dần dần, và cách xử lý blocking code mà không thể async hóa.

Bạn Sẽ Học Được Gì

✅ Khi nào nên migrate (và khi nào không)
✅ Mental model shift từ thread sang coroutine
✅ Pattern refactoring từng bước với code before/after
✅ Cách xử lý blocking library không có async equivalent
✅ Chạy sync và async code song song trong quá trình migrate
✅ Pitfall phổ biến và cách tránh

Prerequisites:
Biết cơ bản về Python threading và quen với cú pháp async/await. Nếu asyncio còn xa lạ, đọc Python Async Programming: Mastering asyncio trước.

1. Có Nên Migrate Không?

Không phải codebase ThreadPoolExecutor nào cũng cần chuyển sang asyncio. Hãy trả lời mấy câu này trước.

Nên migrate nếu:

• Bạn đang làm I/O volume cao (HTTP call, DB query, đọc file) mà phần lớn thời gian là chờ đợi
• Bạn đang chạm giới hạn OS thread hoặc memory overhead quá cao vì hàng trăm thread
• Bạn đang build trên async framework (FastAPI, aiohttp, Starlette) và muốn tích hợp gọn
• Bạn có nhiều task I/O short-lived (ví dụ: fan-out API call)

Không nên migrate nếu:

• Work của bạn là CPU-bound — asyncio không giúp được, multiprocessing mới đúng tool
• Codebase quá synchronous và migration sẽ đụng đến hàng trăm file
• Team chưa có async experience và service không đang bị áp lực gì
• Bạn đang dùng library không có async equivalent và wrap chúng lại còn phức tạp hơn dùng thread

Nói thẳng: ThreadPoolExecutor vẫn ổn với rất nhiều workload. Migrate vì lý do thực tế, không phải vì hype.

2. Mental Model Shift

Trước khi đụng code, phải nắm rõ sự khác biệt này:

Threads: OS quyết định khi nào switch giữa chúng. Code của bạn chạy parallel (trên các core khác nhau với CPU work) hoặc interleave (với I/O). Mỗi thread tốn ~8MB stack memory.

Coroutines: bạn quyết định (ngầm định) khi nào yield. Event loop chạy một coroutine tại một thời điểm, chỉ switch tại các điểm await. Không có OS scheduling overhead, memory mỗi coroutine chỉ vài KB.

Insight chính: concurrency của asyncio đến từ việc cooperative yielding tại các I/O boundary, không phải từ việc chạy code đồng thời. Nếu một coroutine không bao giờ hit await, nó block tất cả mọi thứ.

3. Xác Định Cái Cần Migrate

Bắt đầu bằng cách audit ThreadPoolExecutor usage. Tìm các pattern này:

# Pattern 1: submit + result
with ThreadPoolExecutor(max_workers=10) as pool:
    future = pool.submit(fetch_user, user_id)
    result = future.result()
 
# Pattern 2: map qua một list
with ThreadPoolExecutor(max_workers=20) as pool:
    results = list(pool.map(process_item, items))
 
# Pattern 3: as_completed cho fan-out
with ThreadPoolExecutor(max_workers=50) as pool:
    futures = {pool.submit(call_api, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        data = future.result()

Với mỗi usage, hỏi: function bên dưới có đang làm I/O không?

• requests.get(), urllib.request → có, migrate sang aiohttp/httpx
• psycopg2 query → có, migrate sang asyncpg/psycopg3
• open() + read() cho file lớn → có, migrate sang aiofiles
• subprocess.run() → có, migrate sang asyncio.create_subprocess_exec()
• CPU computation, numpy, image processing → không, giữ nguyên trong thread pool

4. Refactoring Từng Bước

Bước 1: Bắt Đầu từ Leaf Function

Luôn refactor từ dưới lên. Convert các I/O function trong cùng trước, rồi mới ra ngoài.

Before — synchronous với requests:

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import list
 
 
def fetch_user(user_id: int) -> dict:
    response = requests.get(f"https://api.example.com/users/{user_id}")
    response.raise_for_status()
    return response.json()
 
 
def fetch_all_users(user_ids: list[int]) -> list[dict]:
    with ThreadPoolExecutor(max_workers=20) as pool:
        futures = {pool.submit(fetch_user, uid): uid for uid in user_ids}
        results = []
        for future in as_completed(futures):
            results.append(future.result())
    return results

After — async với httpx:

import httpx
import asyncio
from typing import List
 
 
async def fetch_user(client: httpx.AsyncClient, user_id: int) -> dict:
    response = await client.get(f"https://api.example.com/users/{user_id}")
    response.raise_for_status()
    return response.json()
 
 
async def fetch_all_users(user_ids: List[int]) -> List[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [fetch_user(client, uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
    return list(results)

Những thay đổi chính:

• requests → httpx.AsyncClient (cùng API, async-native)
• pool.submit() → tạo coroutine object (task)
• as_completed() → asyncio.gather() để collect tất cả, hoặc asyncio.as_completed() nếu cần kết quả dần dần
• Client được share giữa tất cả các call (connection pooling)

Bước 2: Truyền async Lên Trên

Khi leaf function đã async, caller phải await chúng, nghĩa là caller cũng phải async. Theo chain lên trên.

# Before
def process_batch(batch_ids: list[int]) -> dict:
    users = fetch_all_users(batch_ids)  # trước đây là sync
    return {"count": len(users), "users": users}
 
 
# After
async def process_batch(batch_ids: list[int]) -> dict:
    users = await fetch_all_users(batch_ids)  # giờ là async
    return {"count": len(users), "users": users}

Sự lan truyền này không tránh khỏi — đó là "async infection." Hãy lên kế hoạch cho nó từ đầu. Nếu không thể làm caller async (ví dụ: Django view hoặc CLI entry point), xem Phần 6.

Bước 3: Replace Pattern as_completed

Pattern as_completed từ concurrent.futures có equivalent async trực tiếp:

# Before — thread-based as_completed
from concurrent.futures import ThreadPoolExecutor, as_completed
 
def process_urls(urls: list[str]) -> None:
    with ThreadPoolExecutor(max_workers=30) as pool:
        futures = {pool.submit(scrape, url): url for url in urls}
        for future in as_completed(futures):
            url = futures[future]
            try:
                data = future.result()
                print(f"Done: {url}")
            except Exception as e:
                print(f"Failed {url}: {e}")
 
 
# After — dùng wrapper để giữ metadata
async def process_urls(urls: list[str]) -> None:
    async def scrape_with_url(client: httpx.AsyncClient, url: str) -> tuple[str, dict]:
        data = await scrape(client, url)
        return url, data
 
    async with httpx.AsyncClient() as client:
        tasks = [scrape_with_url(client, url) for url in urls]
        for coro in asyncio.as_completed(tasks):
            try:
                url, data = await coro
                print(f"Done: {url}")
            except Exception as e:
                print(f"Failed: {e}")

Bước 4: Xử Lý Rate Limiting và Concurrency Cap

ThreadPoolExecutor(max_workers=N) tự nhiên giới hạn concurrency. Với asyncio, gather(*tasks) launch tất cả cùng lúc. Với list lớn, bạn cần semaphore:

# Before — max_workers tự giới hạn concurrency
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(call_rate_limited_api, items))
 
 
# After — dùng semaphore
async def call_with_semaphore(
    sem: asyncio.Semaphore,
    client: httpx.AsyncClient,
    item: str
) -> dict:
    async with sem:
        return await call_rate_limited_api(client, item)
 
 
async def process_all(items: list[str]) -> list[dict]:
    sem = asyncio.Semaphore(5)  # tối đa 5 request concurrent
    async with httpx.AsyncClient() as client:
        tasks = [call_with_semaphore(sem, client, item) for item in items]
        return await asyncio.gather(*tasks)

5. Xử Lý Blocking Library

Đây là phần khó. Nhiều Python library — psycopg2, boto3, một số ORM, legacy SDK — là synchronous và không có async equivalent. Bạn có ba lựa chọn:

Option A: Dùng loop.run_in_executor() (Wrap Nhanh)

import asyncio
import boto3
from functools import partial
 
 
s3 = boto3.client("s3")
 
 
async def upload_to_s3(bucket: str, key: str, data: bytes) -> None:
    loop = asyncio.get_event_loop()
    # run_in_executor wrap blocking call trong một thread
    await loop.run_in_executor(
        None,  # None = default ThreadPoolExecutor
        partial(s3.put_object, Bucket=bucket, Key=key, Body=data)
    )

Đây là bridge thực dụng: coroutine của bạn await blocking call mà không block event loop. Call đó vẫn chạy trong thread, nhưng tích hợp gọn với async code.

Set custom executor để kiểm soát thread count:

import asyncio
from concurrent.futures import ThreadPoolExecutor
 
executor = ThreadPoolExecutor(max_workers=10)
 
async def some_async_function():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, blocking_function, arg1, arg2)
    return result

Option B: Chuyển Sang Async Library

Option C: Tách Sync Code Vào Worker Process

Với CPU-heavy + blocking sync work nặng, dùng ProcessPoolExecutor với run_in_executor. Mỗi process có GIL riêng và event loop không bị ảnh hưởng:

from concurrent.futures import ProcessPoolExecutor
import asyncio
 
process_pool = ProcessPoolExecutor(max_workers=4)
 
async def run_cpu_bound(data: bytes) -> bytes:
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(process_pool, heavy_cpu_function, data)
    return result

6. Bridging Sync và Async Trong Quá Trình Migration

Migration hiếm khi là rewrite một ngày. Sẽ có giai đoạn một số caller là sync và một số là async. Đây là cách bridge:

Gọi async từ sync code

import asyncio
 
 
async def async_main() -> None:
    result = await fetch_all_users([1, 2, 3])
    print(result)
 
 
# Entry point — sync context
if __name__ == "__main__":
    asyncio.run(async_main())

Với library hoặc utility cần expose sync interface lên trên async internals:

import asyncio
from typing import TypeVar, Coroutine, Any
 
T = TypeVar("T")
 
 
def run_sync(coro: Coroutine[Any, Any, T]) -> T:
    """Chạy coroutine synchronously. Chỉ dùng tại sync/async boundary."""
    return asyncio.run(coro)
 
 
# Dùng tiết kiệm — chỉ tại boundary thực sự, không chôn trong business logic
result = run_sync(fetch_all_users([1, 2, 3]))

Đừng bao giờ gọi asyncio.run() bên trong event loop đang chạy. Đây là lỗi phổ biến khi gọi async code từ Jupyter notebook hoặc từ bên trong FastAPI endpoint. Dùng await thay thế.

Gọi sync từ async code (không block)

import asyncio
from concurrent.futures import ThreadPoolExecutor
 
_executor = ThreadPoolExecutor(max_workers=10)
 
 
async def call_sync_safely(sync_fn, *args):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(_executor, sync_fn, *args)
 
 
# Usage
async def handler():
    # Đừng làm thế này — block event loop!
    # result = some_sync_library_call()
 
    # Làm thế này
    result = await call_sync_safely(some_sync_library_call, arg1, arg2)
    return result

7. Pitfall Phổ Biến

Pitfall 1: Blocking Event Loop

Bug async phổ biến nhất. Bất kỳ sync I/O hay CPU-heavy code nào trong coroutine đều block tất cả:

# Bad — block toàn bộ event loop
async def bad_handler():
    time.sleep(2)            # block
    data = requests.get(url) # block
    result = json.loads(data) # ổn, CPU work nhanh
 
# Good
async def good_handler():
    await asyncio.sleep(2)           # yield
    async with httpx.AsyncClient() as client:
        response = await client.get(url)  # yield
    result = response.json()              # ổn

Dùng asyncio.get_event_loop().set_debug(True) khi dev — nó log coroutine nào chạy quá 100ms mà không yield.

Pitfall 2: Tạo Event Loop Mới Cho Mỗi Request

# Bad — tạo event loop mới mỗi lần, expensive và error-prone
def handle_request(data):
    loop = asyncio.new_event_loop()
    result = loop.run_until_complete(process(data))
    loop.close()
    return result
 
# Good — dùng asyncio.run() ở top level, hoặc làm handler async
async def handle_request(data):
    return await process(data)

Pitfall 3: Không Await Task

# Bad — fire-and-forget, exception bị nuốt im lặng
async def bad():
    asyncio.create_task(risky_operation())  # không await
    return "done"
 
# Good — collect và await, hoặc handle exception rõ ràng
async def good():
    task = asyncio.create_task(risky_operation())
    try:
        await task
    except Exception as e:
        logger.error(f"Task failed: {e}")
    return "done"

Pitfall 4: Share Object Không Thread-Safe Giữa Coroutine và Executor

Coroutine chạy trên single thread nên object share giữa chúng nhìn chung an toàn. Nhưng nếu bạn mix run_in_executor với shared state, cần cẩn thận:

# Nguy hiểm nếu db_connection không thread-safe
async def dangerous():
    loop = asyncio.get_event_loop()
    # db_connection được access từ nhiều thread qua executor
    result = await loop.run_in_executor(None, db_connection.query, sql)

Dùng connection pool được thiết kế cho concurrent access, hoặc đảm bảo mỗi thread có connection riêng.

8. Đo Lường Kết Quả

Validate rằng migration thực sự giúp ích trước khi tuyên bố thành công:

import asyncio
import time
import httpx
 
 
async def benchmark_async(urls: list[str]) -> float:
    start = time.perf_counter()
    async with httpx.AsyncClient() as client:
        await asyncio.gather(*[client.get(url) for url in urls])
    return time.perf_counter() - start
 
 
def benchmark_threads(urls: list[str]) -> float:
    import requests
    from concurrent.futures import ThreadPoolExecutor
    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=len(urls)) as pool:
        list(pool.map(requests.get, urls))
    return time.perf_counter() - start

Kết quả điển hình với I/O-bound work:

• Threads: scale tốt đến ~100 concurrent task, memory tăng tuyến tính
• Asyncio: scale đến hàng nghìn concurrent task, memory giữ flat

Lợi thế thực sự không phải tốc độ raw — mà là resource efficiency ở scale lớn. Nếu bạn chỉ cần 10 concurrent request, sự khác biệt không đáng kể.

9. Migration Checklist

Dùng cái này khi lên kế hoạch migrate:

Trước khi bắt đầu:
✅ Tìm tất cả ThreadPoolExecutor / ThreadPool usage bằng grep
✅ Phân loại: I/O-bound (migrate) vs CPU-bound (giữ nguyên trong pool)
✅ Kiểm tra xem có async equivalent cho mọi library bạn đang dùng không
✅ Quyết định scope migration (full rewrite vs incremental)

Trong quá trình migrate:
✅ Bắt đầu từ leaf I/O function, truyền async lên trên
✅ Replace sync HTTP/DB library bằng async equivalent
✅ Thêm semaphore ở bất cứ đâu max_workers đang giới hạn concurrency
✅ Wrap blocking call không tránh được bằng run_in_executor
✅ Bật asyncio debug mode trong quá trình development

Sau khi migrate:
✅ Benchmark: so sánh throughput và memory vs trước
✅ Test error handling — async exception surface khác với sync
✅ Kiểm tra fire-and-forget task có unhandled exception không
✅ Review timeout handling — asyncio có asyncio.wait_for()

Tổng Kết

Migrate từ ThreadPoolExecutor sang asyncio là upgrade có ý nghĩa cho Python service nặng I/O, nhưng không miễn phí. Các bước chính:

1. Audit thread pool usage và phân loại I/O vs CPU work
2. Refactor bottom-up — leaf function trước, caller sau
3. Replace sync library bằng async equivalent khi có thể
4. Wrap blocking code bằng run_in_executor khi không có async library
5. Thêm semaphore để replace concurrency cap mà max_workers đã cung cấp
6. Đo lường trước khi tuyên bố thành công

Migration đáng làm khi bạn đang chạy hàng trăm concurrent I/O operation và bắt đầu thấy thread exhaustion, memory cao, hoặc throughput kém dưới load. Với workload nhẹ hơn, ThreadPoolExecutor vẫn là tool hoàn toàn ổn.

Đọc Thêm:
Python Async Programming: Mastering asyncio
Async Processing in FastAPI