Programming

Python Asyncio Concurrency Guide

📅 2026-02-09 | โดย อ.บอม กิตติทัศน์ เจริญพนาสิทธิ์ — SiamCafe.net Since 1997

Python Asyncio Concurrency Guide คืออะไร / ทำไมถึงสำคัญ

น้องๆ เคยเจอไหม โปรแกรมเราทำงานช้ามากกกกกก รันทีนึงกาแฟหมดไปสามแก้ว สมัยผมทำร้านเน็ตนี่ลูกค้าบ่นกระจาย ถ้าเว็บโหลดช้า ลูกค้าก็หนีไปเล่นร้านอื่นหมด Asyncio นี่แหละพระเอกขี่ม้าขาวมาช่วยเราได้เยอะเลย

Asyncio คือ library ที่ช่วยให้ Python ทำงานแบบ concurrency หรือการทำงานหลายอย่างพร้อมๆ กันได้ดีขึ้นมากๆ คิดง่ายๆ เหมือนมีคนๆ เดียวที่เก่งมากๆ ทำหลายอย่างสลับกันไปมาได้เร็วสุดๆ แทนที่จะต้องรอให้งานนึงเสร็จก่อนถึงจะเริ่มอีกงานได้

ทำไมถึงสำคัญน่ะเหรอ? ลองนึกภาพว่าเราต้องโหลดข้อมูลจากหลายๆ เว็บพร้อมกัน ถ้าเราเขียนโปรแกรมแบบปกติ คือรอเว็บแรกโหลดเสร็จก่อนแล้วค่อยโหลดเว็บสอง... กว่าจะโหลดครบทุกเว็บนี่ชาติหน้าเลยนะ แต่ถ้าใช้ Asyncio เราสามารถสั่งให้โหลดทุกเว็บพร้อมๆ กันได้เลย เร็วขึ้นแบบเห็นได้ชัด!

พื้นฐานที่ต้องรู้

ก่อนจะไปลุย Asyncio กันแบบเต็มตัว เราต้องรู้พื้นฐานเหล่านี้ก่อนนะ จะได้ไม่ งง

Coroutines

Coroutine คือ ฟังก์ชันพิเศษที่สามารถหยุด (pause) และกลับมาทำงานต่อ (resume) ได้ ตอนที่มันหยุดทำงานนี่แหละ ที่โปรแกรมสามารถไปทำงานอื่นก่อนได้ คล้ายๆ เราอ่านหนังสือแล้วมีคนมาถามทาง เราก็คั่นหน้าไว้ก่อน พอตอบเสร็จก็กลับมาอ่านต่อที่เดิม

import asyncio

async def my_coroutine():
    print("เริ่มทำงาน!")
    await asyncio.sleep(1)  # จำลองการรอ
    print("ทำงานเสร็จ!")

async def main():
    await my_coroutine()

asyncio.run(main())

ใน code snippet นี้ my_coroutine เป็น coroutine ที่ใช้ await เพื่อหยุดการทำงานชั่วคราวระหว่างรอ asyncio.sleep(1)

Event Loop

Event Loop คือ ตัวจัดการหลักที่คอยวนลูปดูว่ามี coroutine ไหนพร้อมทำงานบ้าง แล้วก็สลับไปทำงาน coroutine นั้นๆ เหมือนเป็น Traffic Controller คอยจัดการจราจรให้รถ (coroutines) วิ่งได้อย่างราบรื่น

สมัยผมทำร้านเน็ต ก็ต้องคอยดู traffic ในร้านตลอดเวลา Event Loop ก็เหมือนกันเลย ต้องคอยดูว่ามีอะไรต้องจัดการบ้าง

async และ await

async คือ keyword ที่ใช้บอกว่าฟังก์ชันนี้เป็น coroutine นะ ส่วน await ใช้เพื่อรอให้ coroutine อื่นทำงานเสร็จก่อน อย่าลืมว่าเราต้องใช้ await ภายใน coroutine เท่านั้นนะ

async def fetch_data(url):
    print(f"กำลังโหลดข้อมูลจาก {url}...")
    # จำลองการโหลดข้อมูล
    await asyncio.sleep(2)
    print(f"โหลดข้อมูลจาก {url} เสร็จแล้ว!")
    return f"ข้อมูลจาก {url}"

async def main():
    task1 = asyncio.create_task(fetch_data("https://www.google.com"))
    task2 = asyncio.create_task(fetch_data("https://www.facebook.com"))

    data1 = await task1
    data2 = await task2

    print(f"ข้อมูลที่ได้: {data1}, {data2}")

asyncio.run(main())

Code นี้แสดงการใช้ async และ await เพื่อโหลดข้อมูลจากสอง URLs พร้อมกัน

วิธีใช้งาน / เริ่มต้นยังไง

มาเริ่มใช้งาน Asyncio กันเลย! ไม่ยากอย่างที่คิด

ขั้นตอนปฏิบัติจริง

สร้าง Coroutine

อันดับแรก เราต้องสร้าง coroutine ขึ้นมาก่อน โดยใช้ keyword async

async def my_task(task_id):
    print(f"Task {task_id}: เริ่มต้น")
    await asyncio.sleep(1)  # จำลองการทำงาน
    print(f"Task {task_id}: เสร็จสิ้น")

สร้าง Task

Task คือตัวแทนของ coroutine ที่กำลังทำงานอยู่ เราใช้ asyncio.create_task() เพื่อสร้าง task จาก coroutine

task1 = asyncio.create_task(my_task(1))
task2 = asyncio.create_task(my_task(2))

รัน Event Loop

สุดท้าย เราต้องรัน event loop เพื่อให้ coroutine ทำงาน โดยใช้ asyncio.run()

async def main():
    task1 = asyncio.create_task(my_task(1))
    task2 = asyncio.create_task(my_task(2))
    await asyncio.gather(task1, task2) # รัน task พร้อมกัน

asyncio.run(main())

asyncio.gather() ใช้รัน tasks หลายๆ ตัวพร้อมกัน และรอให้ทุก task เสร็จสิ้น

เปรียบเทียบกับทางเลือกอื่น

Asyncio ไม่ใช่ทางเลือกเดียวในการทำ concurrency นะ ยังมี threading และ multiprocessing อีกด้วย แต่ละแบบก็มีข้อดีข้อเสียต่างกันไป

Threading เหมาะกับงานที่ต้องรอ I/O (เช่น การอ่านเขียนไฟล์ หรือการติดต่อ network) แต่ไม่เหมาะกับงานที่ต้องคำนวณหนักๆ เพราะติดปัญหา Global Interpreter Lock (GIL)

Multiprocessing เหมาะกับงานที่ต้องคำนวณหนักๆ เพราะแต่ละ process มี memory เป็นของตัวเอง ไม่ติด GIL แต่ก็มี overhead มากกว่า threading

Asyncio ก็เหมาะกับงานที่ต้องรอ I/O เหมือนกัน แต่ใช้ resource น้อยกว่า threading และ multiprocessing ทำให้ scale ได้ดีกว่า

Feature	Threading	Multiprocessing	Asyncio
Concurrency Model	Thread-based	Process-based	Coroutine-based
GIL Limitation	Yes	No	No
Memory Overhead	Low	High	Low
I/O Bound Tasks	Good	Okay	Excellent
CPU Bound Tasks	Poor	Good	Poor
Complexity	Medium	Medium	Medium

จากตารางนี้ น้องๆ จะเห็นว่า Asyncio เหมาะกับงานที่ต้องรอ I/O และต้องการ scale ได้ดี แต่ถ้าเป็นงานที่ต้องคำนวณหนักๆ Multiprocessing จะเป็นทางเลือกที่ดีกว่า ลองเลือกใช้ให้เหมาะสมกับงานนะ

ถ้าอยากอ่านบทความดีๆ แบบนี้อีก แวะมาที่ SiamCafe Blog ได้เลยนะ มีอะไรให้อ่านเยอะแยะ

สมัยผมทำร้านเน็ต ก็ต้องคอยเลือกเทคโนโลยีที่เหมาะสมกับงานเสมอ Asyncio ก็เป็นอีกหนึ่งเครื่องมือที่ช่วยให้เราทำงานได้เร็วขึ้น และประหยัด resource มากขึ้น

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

🎬 วิดีโอแนะนำ

ดูวิดีโอเพิ่มเติมเกี่ยวกับPython Asyncio Concurrency Gui:

Best Practices / เคล็ดลับจากประสบการณ์

เอาล่ะน้องๆ มาถึงตรงนี้ แสดงว่าเริ่มอยากจะลองเล่น Asyncio จริงจังแล้วใช่ไหม? สมัยผมทำร้านเน็ตนี่ concurrency คือหัวใจเลยนะ จะทำยังไงให้เครื่อง Server รับลูกค้าได้เยอะที่สุด smooth ที่สุด นี่คือเคล็ดลับที่ผมอยากแชร์ จากประสบการณ์ตรงเลยนะ

เทคนิคที่ 1: อย่า Blocking เด็ดขาด!

อันนี้สำคัญมากกกก ดอกจันไว้สิบดอก! Asyncio มันเก่งเรื่องสลับ task ไปมาได้ แต่ถ้า task ไหนไปเจอ I/O ที่มัน block (เช่น อ่านไฟล์ใหญ่ๆ หรือ query database แบบ sync) งานเข้าเลยนะ! Event Loop จะหยุดรอ task นั้น task เดียว ทำเอาระบบอืดทั้งระบบ ทางแก้คือใช้ Library ที่มัน support Async I/O เช่น aiohttp (สำหรับ HTTP request) หรือ aiopg (สำหรับ PostgreSQL)

import asyncio
import aiohttp

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["https://www.google.com", "https://www.youtube.com", "https://siamcafe.net/blog/"]
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks) # ใช้ asyncio.gather เพื่อรัน tasks พร้อมๆ กัน
    print(results)

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

สังเกตว่าเราใช้ aiohttp.ClientSession() แทนที่จะใช้ requests แบบเดิมๆ เพราะ aiohttp มัน async ไงล่ะ!

เทคนิคที่ 2: ใช้ asyncio.gather ให้เป็น

asyncio.gather() นี่เหมือนเป็นพระเอกของเราเลยนะ มันช่วยให้เรา run tasks หลายๆ tasks พร้อมๆ กันได้ โดยไม่ต้องเขียนโค้ดเองให้ยุ่งยาก สมัยก่อนผมเขียนเองนะ ปวดหัวมาก! แถมมันยัง return ผลลัพธ์ของแต่ละ task กลับมาให้เราด้วย เรียงตามลำดับที่เราใส่เข้าไปเลย

import asyncio

async def my_task(name, delay):
    await asyncio.sleep(delay)
    print(f"Task {name} done after {delay} seconds")
    return f"Result from {name}"

async def main():
    tasks = [my_task("A", 2), my_task("B", 1), my_task("C", 3)]
    results = await asyncio.gather(*tasks)
    print(f"All tasks done. Results: {results}")

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

ลองรันดู จะเห็นว่า task B เสร็จก่อน เพราะ delay น้อยกว่า แต่ asyncio.gather จะรอให้ทุก task เสร็จก่อน แล้วค่อย return ผลลัพธ์ออกมา

เทคนิคที่ 3: Handle Exceptions ให้ดี

เรื่อง exception นี่สำคัญมาก โดยเฉพาะเวลาที่เรา run tasks พร้อมๆ กัน เพราะถ้า task ไหนเกิด error ขึ้นมา มันอาจจะทำให้โปรแกรมเรา crash ได้ ทางที่ดีคือใช้ try...except ในแต่ละ task เพื่อ handle error ที่อาจจะเกิดขึ้น

import asyncio

async def risky_task(name):
    try:
        result = 10 / 0  # จะเกิด ZeroDivisionError แน่นอน
        return result
    except ZeroDivisionError as e:
        print(f"Task {name} failed with error: {e}")
        return None  # หรือจะ raise exception อีกครั้งก็ได้

async def main():
    tasks = [risky_task("Task 1"), risky_task("Task 2")]
    results = await asyncio.gather(*tasks)
    print(f"Results: {results}")

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

จะเห็นว่าถึงแม้ risky_task จะเกิด error แต่โปรแกรมเราก็ยัง run ต่อไปได้ เพราะเรา handle exception ไว้แล้ว

เทคนิคที่ 4: ใช้ Graceful Shutdown

เวลาที่เราจะปิดโปรแกรมที่ใช้ Asyncio สิ่งสำคัญคือต้องรอให้ tasks ที่กำลัง run อยู่เสร็จก่อน ไม่งั้นข้อมูลอาจจะเสียหายได้ เราสามารถใช้ asyncio.wait_for() เพื่อกำหนด timeout ให้แต่ละ task ได้

import asyncio

async def long_running_task(name):
    try:
        await asyncio.sleep(5)  # สมมติว่าเป็น task ที่ใช้เวลานาน
        print(f"Task {name} finished")
    except asyncio.CancelledError:
        print(f"Task {name} cancelled")

async def main():
    task = asyncio.create_task(long_running_task("Important Task"))

    try:
        await asyncio.wait_for(task, timeout=2) # รอ 2 วินาที
    except asyncio.TimeoutError:
        print("Task timed out, cancelling...")
        task.cancel() # ยกเลิก task
        await asyncio.wait( [task] ) # รอให้ task ยกเลิกตัวเองให้เรียบร้อย

    print("Exiting gracefully")

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

ถ้า task ใช้เวลานานเกิน 2 วินาที มันจะถูก cancel และเราจะออกจากโปรแกรมอย่างสวยงาม

FAQ คำถามที่พบบ่อย

Asyncio เหมาะกับงานแบบไหน?

Asyncio เหมาะกับงานที่เน้น I/O เป็นหลัก เช่น Web Server, API Server, Network programming เพราะมันช่วยให้เราจัดการ connections ได้เยอะขึ้น โดยไม่ต้องสร้าง thread จำนวนมาก สมัยผมทำร้านเน็ตนี่ใช้ Asyncio กับ Server เก็บ Log ไฟล์ คือสบายเลย

Asyncio เร็วกว่า Threading จริงหรือ?

มันตอบยากนะน้อง มันขึ้นอยู่กับลักษณะงาน ถ้างานของเราเป็น CPU-bound (เช่น คำนวณเลขเยอะๆ) Threading อาจจะเร็วกว่า (ถ้าเราใช้ multiprocessing) แต่ถ้าเป็น I/O-bound Asyncio จะได้เปรียบกว่า เพราะมันไม่ต้องเสียเวลาในการสลับ context ระหว่าง thread

Asyncio เรียนยากไหม?

ช่วงแรกๆ อาจจะงงๆ หน่อย เพราะมันมี concepts ใหม่ๆ เยอะ แต่ถ้าเราเข้าใจหลักการทำงานของ Event Loop และ Coroutines แล้ว มันก็ไม่ได้ยากอย่างที่คิด ลองเริ่มจากตัวอย่างง่ายๆ ก่อน แล้วค่อยๆ ขยับไปทำโปรเจกต์ที่ซับซ้อนขึ้น

Error "RuntimeWarning: coroutine '...' was never awaited" คืออะไร?

Error นี้หมายความว่าเราเรียก coroutine function แต่เราไม่ได้ await มัน ทำให้ coroutine ไม่ได้ถูก execute จริงๆ วิธีแก้คือใส่ await หน้า coroutine function หรือใช้ asyncio.create_task() เพื่อสร้าง task แล้วค่อย await task นั้น

สรุป

Asyncio เป็นเครื่องมือที่ทรงพลังในการเขียน Concurrent applications ด้วย Python แต่ก็ต้องทำความเข้าใจ concepts และ best practices ให้ดี ถึงจะใช้งานได้อย่างมีประสิทธิภาพ SiamCafe Blog มีบทความดีๆ เกี่ยวกับ Python อีกเยอะ ลองเข้าไปอ่านดูนะ แล้วอย่าลืมแวะไป iCafeForex ด้วยนะจ๊ะ