การทำงานแบบอะซิงโครนัสด้วย asyncio: coroutine, await, gather

ทำความเข้าใจ concurrency แบบร่วมมือกันของ asyncio: coroutine แบบ lazy, จุดหยุดพัก, gather, TaskGroup และกับดักของการเรียกแบบบล็อก

10 นาที · 3 คำถาม

asyncio ทำให้เกิด concurrency แบบร่วมมือกัน บนเธรดเดียว โดยมี event loop คอยจัดวางลำดับของ coroutine ที่ยอมคืนการควบคุมด้วยความสมัครใจในทุก await จุดพื้นฐานที่มักถูกเข้าใจผิดคือ การเรียกฟังก์ชัน async def ไม่ได้รันอะไรเลย มันสร้างอ็อบเจกต์ coroutine ที่นิ่งเฉย ซึ่งเนื้อในของมันจะเดินหน้าก็ต่อเมื่อถูกรอด้วย await หรือถูกจัดตารางเป็นทาสก์บนลูปเท่านั้น

import asyncio

async def greet() -> str:
    return "hello"

coro = greet()
print(type(coro).__name__)

async def main() -> None:
    print(await coro)

asyncio.run(main())

await ทำเครื่องหมาย จุดหยุดพัก coroutine จะหยุดนิ่งอยู่ตรงนั้น ลูปยึดการควบคุมกลับคืนและเดินหน้าทาสก์อื่นที่พร้อมแล้ว ระหว่าง await สองตัว โค้ดของคุณรันโดยไม่ถูกขัดจังหวะ - ไม่ต้องใช้ล็อกสำหรับโครงสร้างส่วนใหญ่ หากต้องการเปิดใช้งาน coroutine หลายตัวขนานกัน (ในความหมายของ I/O) ให้ห่อพวกมันไว้ใน asyncio.gather การรอหนึ่งวินาทีสามครั้งจะจบภายในหนึ่งวินาทีรวม ไม่ใช่สามวินาที

import asyncio
import time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"

async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("a", 1.0),
        fetch("b", 1.0),
        fetch("c", 1.0),
    )
    elapsed = time.perf_counter() - start
    print(results)
    print(f"total: {elapsed:.1f} s")

asyncio.run(main())

โดยค่าเริ่มต้น gather จะปล่อย exception ตัวแรกที่ถูกโยนให้แพร่กระจายออกมา แต่ยังปล่อยให้ทาสก์ตัวอื่นทำงานต่อไปเบื้องหลัง - ซึ่งไม่ค่อยใช่สิ่งที่คุณต้องการ ด้วย return_exceptions=True exception จะถูกคืนกลับมาในรูปของผลลัพธ์ เพื่อให้คุณไปคัดแยกทีหลัง ตั้งแต่ Python 3.11 เป็นต้นมา ควรเลือกใช้ asyncio.TaskGroup เมื่อเกิดข้อผิดพลาดแรก ทาสก์พี่น้องทั้งหมดจะถูก ยกเลิก อย่างสะอาด และ exception หลายตัวจะลอยขึ้นมารวมกลุ่มกันใน ExceptionGroup เพื่อให้ดักจับด้วย except*

import asyncio
import time

def blocking_io() -> str:
    time.sleep(1)
    return "report written"

async def main() -> None:
    results = await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1, result="tick"),
    )
    print(results)

asyncio.run(main())

ทดสอบความรู้

ตรวจสอบว่าคุณจำประเด็นสำคัญของบทเรียนนี้ได้ครบถ้วน

1. การเรียก fetch('a', 1.0) ที่เขียนไว้โดด ๆ โดยไม่มี await หรือ create_task ให้ผลลัพธ์อะไร?
   • อ็อบเจกต์ coroutine ที่นิ่งเฉย ไม่มีสักบรรทัดในเนื้อของฟังก์ชันที่รัน
   • การรัน coroutine ทันทีแบบซิงโครนัส
   • ทาสก์ที่ถูกจัดตารางบน event loop โดยอัตโนมัติ
   • SyntaxError เพราะ await เป็นสิ่งบังคับ
2. asyncio.sleep(1) สามตัวที่ถูกเปิดใช้งานพร้อมกันผ่าน asyncio.gather ใช้เวลารวมเท่าไร?
   • ประมาณ 3 วินาที เพราะมีเธรดเดียวที่รันโค้ด
   • ทำนายไม่ได้ถ้าไม่รู้จำนวนคอร์
   • ประมาณ 1 วินาที การรอซ้อนทับกัน ลูปสลับจากทาสก์หนึ่งไปอีกทาสก์หนึ่ง
3. ทำไม time.sleep(2) จึงเป็นสิ่งต้องห้ามภายใน coroutine?
   • มันโยน exception ในบริบทแบบ async
   • มันบล็อกเธรดของ event loop ไม่มีทาสก์อื่นเดินหน้าได้เป็นเวลา 2 วินาที
   • มันแค่แม่นยำน้อยกว่า asyncio.sleep
   • มันใช้หน่วยความจำเป็นสองเท่า