โครูทีนและลูปเหตุการณ์

การทำงานพร้อมกันคืออะไร

# Traditional: blocking I/O holds the whole thread
# Asyncio: while waiting for I/O, run other coroutines

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)   # yield control
    print("World")

asyncio.run(main())

async def — ฟังก์ชันโครูทีน

import asyncio

async def greet(name):
    return f"Hello, {name}"

result = asyncio.run(greet("Alice"))
print(result)   # Hello, Alice

ลูปเหตุการณ์

import asyncio

async def compute():
    await asyncio.sleep(0)
    return 42

value = asyncio.run(compute())
print(value)   # 42

# For advanced use:
loop = asyncio.new_event_loop()
result = loop.run_until_complete(compute())
loop.close()

await หยุดการทำงานชั่วคราว

import asyncio

async def task(name, delay):
    print(f"{name} starting")
    await asyncio.sleep(delay)
    print(f"{name} done")

async def main():
    await task("A", 2)
    await task("B", 1)   # sequential here

asyncio.run(main())

asyncio.sleep()

import asyncio

async def heartbeat():
    while True:
        print("ping")
        await asyncio.sleep(1)

# asyncio.run(heartbeat())  # runs forever

asyncio.run() จุดเริ่มต้น

import asyncio

async def main():
    print("running")

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

การตรวจสอบว่ากำลังอยู่ภายในลูปเหตุการณ์หรือไม่

import asyncio

async def inner():
    loop = asyncio.get_running_loop()
    print(f"Loop: {loop}")

asyncio.run(inner())

การเรียกใช้โค้ดที่บล็อกการทำงาน

import asyncio, time

def blocking():
    time.sleep(2)
    return "done"

async def main():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, blocking)
    print(result)

asyncio.run(main())

การเชื่อมโยงโครูทีน

import asyncio

async def fetch_data():
    await asyncio.sleep(1)
    return {"value": 42}

async def process():
    data = await fetch_data()
    return data["value"] * 2

async def main():
    result = await process()
    print(result)   # 84

asyncio.run(main())

การระบุโครูทีน

import asyncio

async def my_coro():
    pass

print(asyncio.iscoroutinefunction(my_coro))   # True
coro = my_coro()
print(asyncio.iscoroutine(coro))              # True
coro.close()   # clean up the un-awaited coroutine

โหมดดีบัก asyncio

import asyncio

async def main():
    pass

asyncio.run(main(), debug=True)

# Or:
# PYTHONASYNCIODEBUG=1 python script.py

ตรวจสอบอย่างรวดเร็ว

สรุปทบทวน