asyncio ใน Python: เมื่อการรอไม่ใช่เรื่องน่าเบื่ออีกต่อไป

สวัสดีทุกคน! คือบางทีเนี่ย เราเขียน Python แล้วเจอเคสที่ต้องไปดึงข้อมูลจากหลายๆ ที่พร้อมกัน หรือต้องรอ I/O อะไรบางอย่างนานๆ เช่น ไปเรียก API, อ่านไฟล์ใหญ่ๆ, หรือคุยกับ Database?

ปกติเลย Python มันจะทำงานแบบ "Sync" อะ คือทำทีละบรรทัด รอให้บรรทัดแรกเสร็จก่อนถึงจะไปบรรทัดถัดไปได้ ถ้าบรรทัดแรกมันต้องรอนานๆ เช่น รอเรียก API เป็นสิบวิ ตัวโปรแกรมเราก็จะค้างอยู่อย่างงั้นเลยไง เสียเวลาไปเยอะเลย!

ทำไมต้อง asyncio? ก็เพราะมัน "ไม่รอ" ไง

asyncio เนี่ย มันช่วยให้เราเขียนโค้ดแบบ "Asynchronous" ได้ คือเวลาที่โค้ดเราไปเจออะไรที่ต้อง "รอ" เช่น ไปเรียก API มันจะไม่ยืนรอเฉยๆ แต่มันจะไปทำอย่างอื่นก่อน แล้วพองานนั้นเสร็จค่อยกลับมาทำต่อ นี่แหละคือความเจ๋งของมัน เหมาะกับงานประเภท I/O-bound มากๆ

ลองนึกภาพว่าคุณสั่งของออนไลน์หลายๆ อย่างอะ แทนที่จะรอให้ของชิ้นแรกมาส่งถึงบ้านก่อนแล้วค่อยกดสั่งชิ้นที่สอง asyncio มันเหมือนคุณกดสั่งไปพร้อมๆ กันเลย แล้วค่อยรอรับของทีเดียวอะ เข้าใจง่ายขึ้นมะ?

มาลองเขียน asyncio แบบง่ายๆ กัน

หัวใจของ asyncio คือ async def กับ await

• async def: ใช้ประกาศฟังก์ชันที่สามารถ "ถูกรอ" ได้ หรือฟังก์ชันที่ข้างในมีอะไรที่ต้องรอ
• await: ใช้สำหรับ "รอ" การทำงานของฟังก์ชันที่เป็น async def

ลองดูตัวอย่างนี้นะ ฟังก์ชันแค่รอเฉยๆ

import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay) # อันนี้แหละที่ไปรอ แต่โปรแกรมไม่ค้าง
    print(f"{time.time():.2f}: {what}")

async def main():
    print(f"{time.time():.2f}: Hello!")
    task1 = asyncio.create_task(say_after(1, 'world'))
    task2 = asyncio.create_task(say_after(2, 'everyone'))

    # อันนี้คือรอให้ task1 กับ task2 ทำงานเสร็จ
    await task1
    await task2

    print(f"{time.time():.2f}: Goodbye!")

if __name__ == "__main__":
    # ต้องรันด้วย asyncio.run()
    asyncio.run(main())

# Output ที่ควรจะได้ (เวลาอาจจะไม่เป๊ะนะ และจะรันไปข้างหน้า)
# 1678886400.00: Hello!
# 1678886401.00: world
# 1678886402.00: everyone
# 1678886402.00: Goodbye!

เห็นมะ? world ขึ้นมาก่อน everyone ทั้งๆ ที่ main ไม่ได้รอให้ say_after(1, 'world') จบแล้วค่อยสั่ง say_after(2, 'everyone') แต่มันสั่งไปพร้อมๆ กันเลย พองานไหนเสร็จก่อนก็แสดงผลก่อนนั่นแหละ

ของจริงละนะ: ดึงข้อมูลจากหลาย API พร้อมกัน

คราวนี้มาลองใช้กับงานจริงๆ ที่เจอโคตรบ่อย คือไปดึงข้อมูลจากหลายๆ API endpoint พร้อมกัน ลองใช้ httpx เพราะมันรองรับ asyncio ได้ดี และใช้ง่ายกว่า aiohttp สำหรับงานง่ายๆ (ความเห็นส่วนตัวนะ)

ก่อนอื่นก็ pip install httpx ก่อนเลยนะ

import asyncio
import httpx
import time

async def fetch_user(user_id):
    print(f"[{time.time():.2f}] กำลังดึงข้อมูล user {user_id}...")
    url = f"https://jsonplaceholder.typicode.com/users/{user_id}"
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, timeout=5) # บางทีก็อยากตั้ง timeout นะ กันค้างนานๆ
            response.raise_for_status() # ถ้ามี error พวก 4xx, 5xx จะได้รู้
            user_data = response.json()
            print(f"[{time.time():.2f}] ดึงข้อมูล user {user_id} เสร็จแล้ว: {user_data['name']}")
            return user_data
        except httpx.HTTPStatusError as e:
            print(f"[{time.time():.2f}] Error HTTP สำหรับ user {user_id}: {e.response.status_code} - {e.response.text}")
            return None
        except httpx.RequestError as e:
            print(f"[{time.time():.2f}] Error เชื่อมต่อสำหรับ user {user_id}: {e}")
            return None

async def main():
    start_time = time.time()
    user_ids = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    # สร้าง tasks สำหรับแต่ละ user_id
    tasks = [fetch_user(uid) for uid in user_ids]

    # await asyncio.gather() จะรันทุก tasks พร้อมกัน และรอให้ทุกอันเสร็จ
    results = await asyncio.gather(*tasks)

    print(f"\n[{time.time():.2f}] ทุกอย่างเสร็จสิ้นแล้วในเวลา {time.time() - start_time:.2f} วินาที")

    # ลองปริ้นต์ชื่อผู้ใช้ที่ดึงมาได้
    for user in results:
        if user:
            print(f"- {user['name']}")

if __name__ == "__main__":
    asyncio.run(main())

ถ้าลองรันโค้ดข้างบนเนี่ย คุณจะเห็นว่ามันไม่ได้ดึง user 1 เสร็จแล้วค่อยไปดึง user 2 แต่มันจะสั่งดึง user ทั้งหมดไปพร้อมๆ กัน แล้วพอใครเสร็จก่อนก็ปริ้นต์ออกมาก่อน ทำให้ภาพรวมมันเร็วขึ้นมากๆ ยิ่งมีหลายๆ API ที่ต้องเรียกยิ่งเห็นผลชัดเจนเลย

ข้อควรระวังเล็กๆ น้อยๆ ที่เคยเจอ

ตอนแรกๆ ผมก็เคยงงนะว่า 'เห้ย! ทำไมมันไม่เร็วขึ้นเลยวะ?' ปรากฏว่าลืมใส่ await ตรงที่เรียกฟังก์ชัน หรือบางทีฟังก์ชันที่เรียกไปมันเป็น Sync ปกติ ไม่ได้เป็น async def (เช่น พวกไลบรารีเก่าๆ ที่ไม่ได้รองรับ asyncio) อันนี้ก็ช่วยไม่ได้นะ มันก็ยังรออยู่ดีแหละ

แล้วก็ asyncio เนี่ยมันเหมาะกับงานที่เป็น I/O-bound (งานที่ต้องรอ) ถ้างานของคุณเป็น CPU-bound (งานที่ใช้ CPU เยอะๆ เช่น คำนวณอะไรซับซ้อน) asyncio อาจจะไม่ใช่คำตอบที่ดีที่สุดนะ เพราะมันรันบนเธรดเดียว มันแค่สลับไปทำงานอื่นตอนที่รอ I/O แต่ไม่ได้ทำให้โค้ดที่ใช้ CPU ทำงานเร็วขึ้น ถ้าเป็นงาน CPU-bound แนะนำไปใช้ multiprocessing จะเหมาะกว่า

สรุปนะ

asyncio คือเครื่องมือที่ดีมากสำหรับ Python ในการจัดการงานที่ต้องรอ หรือพวก I/O-bound ต่างๆ ทำให้โปรแกรมเราตอบสนองได้ดีขึ้นมากๆ โดยเฉพาะกับการเขียน Web Server, API Client หรืออะไรที่ต้องคุยกับเครือข่ายเยอะๆ ลองเอาไปใช้ดูนะ แล้วจะรู้สึกว่าชีวิตมันง่ายขึ้นเยอะเลย!

หวังว่าบทความนี้จะพอเป็นประโยชน์ให้ทุกคนได้บ้างนะฮะ ถ้ามีอะไรสงสัยก็ลองหาข้อมูลเพิ่มได้เลย asyncio มีอะไรให้เล่นอีกเยอะ!