Asyncio ใน Python: async/await, TaskGroup, เธรด, multiprocessing และ GIL

Asyncio คือไลบรารีมาตรฐานของ Python สำหรับโค้ดแบบอะซิงโครนัส ฟังก์ชันที่ประกาศด้วย async def สามารถหยุดชั่วคราวที่คำสั่ง await ได้ และระหว่างที่รอการตอบกลับจากเครือข่าย event loop จะทำงานอื่นต่อ ด้วยวิธีนี้ หลายร้อยคำขอจึงทำงานพร้อมกันในเธรดเดียว แต่ asyncio เป็นเพียงหนึ่งในสามเครื่องมือ นอกจากนี้ยังมีเธรด (threading) และโปรเซส (multiprocessing) บทความนี้จะอธิบายการทำงานของ asyncio ใน Python วิธีเลือกเครื่องมือให้เหมาะกับงาน และความเกี่ยวข้องของ GIL

การรอกับการคำนวณเป็นงานคนละประเภท

ก่อนจะเข้าใจส่วนอื่น ต้องแยกให้ได้ก่อนว่าโปรแกรมช้าเพราะเหตุผลใดใน 2 แบบ:

• I/O-bound โปรแกรมกำลังรอ I/O เช่น การตอบกลับจาก API ฐานข้อมูล หรือการอ่านดิสก์ ซีพียูว่างในช่วงนั้น
• CPU-bound โปรแกรมกำลังคำนวณ เช่น ประมวลผลรูปภาพ หรือวนลูปหลายล้านรอบ ซีพียูทำงานเต็มที่

สำหรับการรอ asyncio หรือเธรดเหมาะที่สุด เพราะขณะที่งานหนึ่งรออยู่ งานอื่นก็ทำงานได้ ส่วนการคำนวณต้องใช้การประมวลผลขนานจริงบนหลายคอร์ ซึ่งทำได้ด้วยโปรเซส การสับสนสองกรณีนี้เป็นข้อผิดพลาดหลัก เพราะเธรดจะไม่ทำให้การคำนวณด้วย Python ล้วนเร็วขึ้น ขณะที่การใช้โปรเซสเพื่อส่งคำขอ HTTP สามครั้งก็เหมือนใช้ปืนใหญ่ยิงนก

asyncio คืออะไร: คอรูทีนและ event loop

โค้ดอะซิงโครนัสสร้างจากคำสองคำ async def ประกาศคอรูทีน ซึ่งเป็นฟังก์ชันที่หยุดชั่วคราวได้ การเรียกใช้ไม่ได้รันเนื้อหาทันที แต่สร้างออบเจกต์คอรูทีนขึ้นมา ส่วน await ภายในฟังก์ชันหมายความว่า "ตรงนี้ฉันจะรอ เชิญงานอื่นทำก่อน" ผู้จัดการทั้งหมดคือ event loop ซึ่งเริ่มด้วย asyncio.run()

import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # จำลองการเรียก API
    return {"USD": 92.5, "EUR": 100.1, "CNY": 12.7}[currency]

async def main():
    rate = await fetch_rate("USD")
    print(rate)

asyncio.run(main())  # 92.5

asyncio.sleep() คือการพักแบบอะซิงโครนัส ซึ่งคืนการควบคุมให้ event loop ส่วน time.sleep() ภายในคอรูทีนจะบล็อก loop ทั้งหมด และงานอื่นต้องรอ

ทีละงานหรือพร้อมกัน

หากเขียน await ในลูป คำขอจะทำทีละตัว การดึงสามสกุลเงินจะใช้เวลาสามวินาที หากต้องการให้ทำพร้อมกัน ต้องสั่งให้เป็นงาน (task)

import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # ทีละตัว
    print(f"{time.perf_counter() - start:.1f} วินาที")   # 3.0 วินาที

asyncio.run(main())

TaskGroup: เริ่มพร้อมกันและรอให้ครบ

วิธีสมัยใหม่ (Python 3.11 ขึ้นไป) คือการใช้กลุ่มของงาน ภายในบล็อก async with แต่ละงานจะเริ่มทันทีผ่าน tg.create_task() และการออกจากบล็อกเกิดขึ้นเมื่อทุกงานเสร็จ

async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = {cur: tg.create_task(fetch_rate(cur)) for cur in ("USD", "EUR", "CNY")}
    rates = {cur: t.result() for cur, t in tasks.items()}
    print(rates, f"{time.perf_counter() - start:.1f} วินาที")   # ... 1.0 วินาที

asyncio.run(main())

TaskGroup ทำงานอย่างเข้มงวด: หากงานหนึ่งล้มเหลว งานที่เหลือจะถูกยกเลิก และข้อผิดพลาดทั้งหมดถูกรวมเป็น ExceptionGroup ซึ่งดักจับได้ด้วย except*:

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_rate("USD"))
            tg.create_task(fetch_rate("XXX"))   # KeyError
    except* KeyError as eg:
        print("ไม่รู้จักสกุลเงินนี้:", eg.exceptions)

asyncio.gather และ create_task

ก่อนมี TaskGroup มาตรฐานคือ asyncio.gather() ซึ่งเริ่มคอรูทีนพร้อมกันและคืนผลลัพธ์ตามลำดับเดิม โค้ดเก่าจำนวนมากยังใช้อยู่

async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # ถ้าใส่ return_exceptions=True ข้อผิดพลาดจะถูกส่งกลับในลิสต์แทนที่จะหยุดทั้งหมด

asyncio.create_task() เริ่มคอรูทีนในพื้นหลังโดยไม่มีกลุ่ม ต้องเก็บการอ้างอิงถึงงานไว้ มิฉะนั้นตัวเก็บขยะอาจลบงานนั้นทิ้ง และข้อผิดพลาดจะไม่มีใครเห็น ดังนั้นสำหรับโปรเจกต์ใหม่ TaskGroup จึงปลอดภัยกว่า

การยกเลิกงานและการจับเวลาหมดอายุ

async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # ภายในงานจะเกิด CancelledError
    try:
        await task
    except asyncio.CancelledError:
        print("ถูกยกเลิก")

    try:
        async with asyncio.timeout(0.5):   # Python 3.11 ขึ้นไป
            await fetch_rate("EUR")
    except TimeoutError:
        print("ใช้เวลานานเกินไป")

ในเวอร์ชันเก่า ใช้ asyncio.wait_for(coro, timeout=0.5) แทน asyncio.timeout() นอกจากนี้ยังมี asyncio.wait() ซึ่งคืนเซตของงานที่เสร็จแล้วและยังไม่เสร็จ และสามารถรอเฉพาะงานแรกที่เสร็จได้ด้วย return_when=asyncio.FIRST_COMPLETED

ใน Python 3.15 ตามที่แสดงในคอร์ส TaskGroup มีเมธอด tg.cancel() ซึ่งยกเลิกงานทั้งหมดในกลุ่มโดยไม่ต้องมีข้อยกเว้นพิเศษ ช่วยแก้สถานการณ์ "ใครเสร็จก่อนได้" เช่น ถามบริการสำรองสามแห่ง รับคำตอบแรก แล้วยกเลิกที่เหลือ ก่อนหน้านี้ต้องโยนข้อยกเว้นสัญญาณของตัวเองหรือยกเลิกงานทีละตัวด้วยมือ

การซิงโครไนซ์: Semaphore, Lock, Event, Queue

asyncio มีตัวจัดการของตัวเองที่ทำงานร่วมกับ await ตัวที่มีประโยชน์ที่สุดคือ Semaphore ซึ่งจำกัดจำนวนการทำงานพร้อมกัน เพื่อไม่ให้ยิงคำขอหลักพันใส่ API ของผู้อื่นจนล่ม

sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # ไม่เกิน 10 งานพร้อมกัน
        await asyncio.sleep(0.1) # ตรงนี้จะเป็นคำขอจริง
        return url

• asyncio.Lock ให้เพียงหนึ่งคอรูทีนเข้าไปในส่วนวิกฤต (critical section)
• asyncio.Event ให้คอรูทีนหนึ่งรอด้วย await event.wait() อีกตัวส่งสัญญาณด้วย event.set()
• asyncio.Queue เป็นคิวแบบผู้ผลิต–ผู้บริโภค โดยใช้ await q.put() await q.get() และ q.task_done()

ข้อผิดพลาดที่ต้องระวัง: ลืม await

ข้อผิดพลาดที่พบบ่อยที่สุดคือเรียกคอรูทีนโดยไม่มี await:

async def main():
    rate = fetch_rate("USD")    # ลืม await
    print(rate * 2)             # TypeError
# RuntimeWarning: coroutine 'fetch_rate' was never awaited

Python ไม่ได้รันฟังก์ชัน แต่สร้างออบเจกต์คอรูทีนเท่านั้น หากเห็นข้อความ "was never awaited" ในล็อก ให้มองหา await ที่ขาดไป ข้อผิดพลาดที่พบบ่อยอีกอย่างคือการเรียกที่บล็อกอยู่ เช่น time.sleep หรือ requests.get ภายในคอรูทีน สำหรับ HTTP ให้ใช้ไลบรารีแบบอะซิงโครนัส เช่น httpx หรือ aiohttp และโค้ดที่เป็นแบบซิงโครนัสซึ่งหลีกเลี่ยงไม่ได้ให้ส่งไปทำงานในเธรดด้วย await asyncio.to_thread(func, arg)

การทำงานหลายเธรดใน Python: โมดูล threading

เธรด (thread) คือเส้นทางการทำงานอิสระภายในโปรเซสเดียว และแชร์หน่วยความจำร่วมกัน เธรดทำงานได้ดีมากกับการรอเครือข่าย และไม่ต้องเขียนโค้ดใหม่เป็น async

import threading, time

def download(name):
    time.sleep(1)               # รอเครือข่าย
    print(name, "เสร็จแล้ว")

threads = [threading.Thread(target=download, args=(n,)) for n in ("a", "b", "c")]
for t in threads:
    t.start()
for t in threads:
    t.join()                    # ทั้งหมดใช้เวลาประมาณ 1 วินาที

หน่วยความจำร่วมกันเป็นทั้งข้อดีและแหล่งของข้อผิดพลาด หากเธรดสองตัวแก้ตัวแปรเดียวกัน ต้องใช้ threading.Lock โดยทั่วไปแนะนำให้ใช้พูลแทนการจัดการเธรดเอง

ThreadPoolExecutor

concurrent.futures.ThreadPoolExecutor สร้างพูลของเธรดและกระจายงานให้ เมธอด map คืนผลลัพธ์ตามลำดับต้นฉบับ

from concurrent.futures import ThreadPoolExecutor
import time

def fetch(url):
    time.sleep(1)
    return f"{url}: 200"

urls = [f"https://example.com/{i}" for i in range(5)]
with ThreadPoolExecutor(max_workers=5) as pool:
    for result in pool.map(fetch, urls):
        print(result)           # ทั้งหมดใช้เวลาประมาณ 1 วินาที

GIL คืออะไร และทำไมเธรดไม่ช่วยให้การคำนวณเร็วขึ้น

GIL (Global Interpreter Lock) คือล็อกส่วนกลางของตัวแปลภาษา CPython ในแต่ละช่วงเวลา มีเพียงเธรดเดียวที่รันบายต์โค้ด (bytecode)ของ Python ขณะที่เธรดหนึ่งรอ I/O มันจะปล่อย GIL ทำให้เธรดมีประโยชน์กับเครือข่าย แต่กับการคำนวณล้วน ๆ เธรดสี่ตัวจะไม่เร็วกว่าเธรดเดียว เพราะต่างต้องผลัดกันถือล็อก

from concurrent.futures import ThreadPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

with ThreadPoolExecutor(4) as pool:
    list(pool.map(heavy, [5_000_000] * 4))   # ใกล้เคียงกับทำทีละตัว

ไลบรารีที่เขียนด้วย C (เช่น NumPy และอื่น ๆ) มักปล่อย GIL ระหว่างการดำเนินการหนัก ทำให้เธรดช่วยเพิ่มความเร็วได้กับไลบรารีเหล่านี้

multiprocessing: การประมวลผลขนานจริง

แต่ละโปรเซสมีตัวแปลภาษาและ GIL ของตัวเอง ดังนั้นโปรเซสคำนวณบนคนละคอร์พร้อมกันได้จริง ข้อแลกเปลี่ยนคือหน่วยความจำแยกกัน ข้อมูลที่ส่งระหว่างโปรเซสต้องถูกแปลงด้วยการทำให้เป็นอนุกรม (pickle) และการเริ่มโปรเซสมีต้นทุนสูงกว่าเธรด

from concurrent.futures import ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":      # จำเป็นบน Windows และ macOS
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))

หากไม่มี if __name__ == "__main__" โปรเซสลูกจะเริ่มสร้างโปรเซสใหม่ตอน import โมดูล นอกจากนี้ยังมีโมดูลระดับต่ำ multiprocessing ที่มี Process Pool และ Queue แต่ ProcessPoolExecutor ใช้งานง่ายกว่า และมีอินเทอร์เฟซเหมือนพูลของเธรด

จะเลือก asyncio, threading หรือ multiprocessing

Python ที่ไม่มี GIL และ subinterpreter: ของใหม่

ตั้งแต่ Python 3.13 มีบิลด์ที่ไม่มี GIL (free-threading, PEP 703) ซึ่งระบุด้วยตัวอักษร t ในชื่อ เช่น python3.15t ใน 3.13 บิลด์นี้ยังเป็นการทดลอง ส่วนตั้งแต่ 3.14 (PEP 779) ได้รับการสนับสนุนอย่างเป็นทางการ ในบิลด์นี้เธรดคำนวณขนานบนหลายคอร์ได้จริง แต่ค่าเริ่มต้นที่ได้รับยังเป็น Python ปกติที่มี GIL โค้ดแบบเธรดเดียวในบิลด์ free-threaded จะทำงานช้าลงเล็กน้อย และไลบรารีบางตัวยังไม่รองรับ

import sys
print(sys._is_gil_enabled())   # False ในบิลด์ free-threaded

ทางเลือกอีกทางสำหรับการคำนวณขนานปรากฏใน 3.14 (PEP 734) คือ subinterpreter ในโมดูล concurrent.interpreters ซึ่งมีตัวแปลภาษาที่แยกจากกันหลายตัวในโปรเซสเดียว โดยแต่ละตัวมี GIL ของตัวเอง สะดวกที่สุดเมื่อใช้ผ่าน InterpreterPoolExecutor ซึ่งหน้าตาเหมือนพูลของเธรด จากการทดสอบในคอร์ส งานหนักสี่งานใช้เวลา 0.39 วินาที เทียบกับ 1.03 วินาทีเมื่อทำในเธรดเดียว

สรุป

asyncio ใน Python เป็นเครื่องมือสำหรับงานที่ต้องรอ ได้แก่ คอรูทีนที่ประกาศด้วย async def การหยุดรอด้วย await และการเริ่มงานพร้อมกันผ่าน TaskGroup หรือ gather เธรดแก้ปัญหาเดียวกันโดยไม่ต้องเขียนโค้ดใหม่เป็น async ส่วนการคำนวณหนักต้องใช้โปรเซส เพราะ GIL ไม่ให้เธรดของ CPython ปกติคำนวณขนานกันได้ Free-threading และ subinterpreter กำลังเปลี่ยนภาพนี้ แต่กฎพื้นฐานที่ว่า "รอ ใช้ asyncio หรือเธรด คำนวณ ใช้โปรเซส" ยังใช้ได้อยู่ ในคอร์สหัวข้อนี้ถูกสาธิตแบบลงมือทำ โดยโปรแกรมติดตามค่าใช้จ่ายดึงอัตราแลกเปลี่ยนหลายสกุลพร้อมกันในเวลาหนึ่งวินาทีแทนที่จะเป็นสามวินาที