Asyncio katika Python: async/await, TaskGroup, threads, multiprocessing na GIL

Asyncio ni maktaba ya kawaida ya Python kwa msimbo wa asynchronous. Kazi zilizotangazwa kwa async def zinaweza kusimama kwenye await, na wakati mmoja anasubiri jibu la mtandao, kitanzi cha matukio kinaendesha kazi nyingine. Hivyo maombi mamia yanatekelezwa pamoja kwenye thread moja. Lakini asyncio ni mojawapo tu kati ya zana tatu: kuna pia threads (threading) na michakato (multiprocessing). Hebu tuone jinsi asyncio python inavyofanya kazi, nini cha kuchagua kwa kazi yako, na GIL inahusiana vipi na hili.

Kusubiri na kuhesabu ni mambo tofauti

Bila tofauti hii hakuna kitakachoeleweka zaidi. Programu inachelewa kwa sababu moja kati ya mbili:

• I/O-bound — inasubiri ingizo-pato: jibu la API, hifadhidata, kusoma diski. Wakati huo processor haifanyi kazi.
• CPU-bound — inahesabu: inachakata picha, inapitia mizunguko ya mamilioni ya rekodi. Processor imejaa.

Kwa kusubiri zinafaa asyncio au threads: wakati operesheni moja inasubiri, nyingine inatekelezwa. Kwa hesabu halisi sambamba kwenye cores kadhaa zinahitajika michakato. Kuchanganya haya ndiko kosa kubwa zaidi: threads hazitaharakisha hesabu kwenye Python safi, na michakato kwa ajili ya maombi matatu tu ni kutumia bunduki kuua nzi.

Asyncio ni nini: coroutines na kitanzi cha matukio

Msimbo wa asynchronous umejengwa juu ya maneno mawili. async def inatangaza coroutine — kazi inayoweza kusimama. Wito wake haitekelezi mwili, bali unaunda kitu cha coroutine. await ndani yake inasema: "hapa nitasubiri, na wewe endelea na kazi nyingine". Yote yanasimamiwa na kitanzi cha matukio (event loop), kinachoanzishwa kwa asyncio.run().

import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # kuiga ombi kwa 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() ni pumziko la asynchronous: linakabidhi udhibiti kwa kitanzi cha matukio. time.sleep() ya kawaida ndani ya coroutine ingezuia kitanzi kizima, na kazi nyingine zingesimama.

Mfululizo au kwa pamoja

Ukiandika await ndani ya mzunguko, maombi yatapita moja baada ya jingine — sarafu tatu zitachukua sekunde tatu. Ili yapite kwa pamoja, yanapaswa kuanzishwa kama kazi.

import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # moja baada ya nyingine
    print(f"{time.perf_counter() - start:.1f} s")   # 3.0 s

asyncio.run(main())

TaskGroup: anzisha pamoja na usubiri wote

Njia ya kisasa (Python 3.11+) ni kikundi cha kazi. Ndani ya kizuizi cha async with kila kazi inaanza mara moja kupitia tg.create_task(), na kutoka kwenye kizuizi hutokea wakati kazi zote zimeisha.

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} s")   # ... 1.0 s

asyncio.run(main())

TaskGroup ina msimamo mkali: kazi moja ikianguka, nyingine zinaghairiwa, na makosa yote yanakusanywa kwenye ExceptionGroup. Yanakamatwa kwa 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("sarafu isiyojulikana:", eg.exceptions)

asyncio.gather na create_task

Kabla ya TaskGroup, kiwango kilikuwa asyncio.gather(): inaanzisha coroutines kwa pamoja na kurudisha matokeo kwa mpangilio uleule. Bado kuna msimbo mwingi wa aina hii.

async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # na return_exceptions=True makosa yanarudi kwenye orodha, bila kusimamisha yote

asyncio.create_task() inaanzisha coroutine chinichini bila kikundi. Marejeo ya kazi lazima yahifadhiwe, vinginevyo mkusanyaji taka anaweza kuifuta, na makosa yake hayataonekana kwa mtu yeyote. Kwa hiyo kwa miradi mipya TaskGroup ni salama zaidi.

Kughairi kazi na muda wa juu

async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # ndani ya kazi itatokea CancelledError
    try:
        await task
    except asyncio.CancelledError:
        print("imeghairiwa")

    try:
        async with asyncio.timeout(0.5):   # Python 3.11+
            await fetch_rate("EUR")
    except TimeoutError:
        print("imechukua muda mrefu sana")

Katika matoleo ya zamani badala ya asyncio.timeout() hutumika asyncio.wait_for(coro, timeout=0.5). Pia kuna asyncio.wait() — inarudisha seti za kazi zilizokamilika na zisizokamilika, na inaweza kusubiri ya kwanza (return_when=asyncio.FIRST_COMPLETED).

Katika Python 3.15, kama inavyoonyeshwa kwenye kozi, TaskGroup imepata mbinu tg.cancel(): inaghairi kazi zote za kikundi bila ubaguzi. Hii inatatua hali ya "nani wa kwanza": kuuliza huduma tatu za kioo, kuchukua jibu la kwanza na kughairi zilizobaki. Awali ilihitaji kutupa exception yako ya ishara au kughairi kazi kwa mkono.

Usawazishaji: Semaphore, Lock, Event, Queue

Asyncio ina zana zake zinazofanya kazi kwa await. Muhimu zaidi ni Semaphore: inapunguza idadi ya operesheni zinazofanyika kwa wakati mmoja, ili usiizidishe API ya mtu mwingine kwa maombi elfu.

sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # si zaidi ya 10 kwa wakati mmoja
        await asyncio.sleep(0.1) # hapa kungekuwa na ombi
        return url

• asyncio.Lock — coroutine moja tu katika sehemu muhimu.
• asyncio.Event — coroutine moja inasubiri await event.wait(), nyingine inatoa ishara kwa event.set().
• asyncio.Queue — foleni ya "mzalishaji — mtumiaji": await q.put(), await q.get(), q.task_done().

Mitego: umesahau await

Kosa la kawaida zaidi ni kuita coroutine bila await:

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

Python haikutekeleza kazi, ilitengeneza tu kitu cha coroutine. Ukiona "was never awaited" kwenye logi, tafuta await iliyokosekana. Kosa la pili la kawaida ni wito unaozuia (time.sleep, requests.get) ndani ya coroutine. Kwa HTTP tumia maktaba za asynchronous (httpx, aiohttp), na msimbo wa synchronous usioepukika uhamishe kwenye thread kwa await asyncio.to_thread(func, arg).

Multithreading katika Python: moduli threading

Thread ni mstari huru wa utekelezaji ndani ya mchakato mmoja, threads zinashiriki kumbukumbu ya pamoja. Kwa kusubiri mtandao zinafanya kazi vizuri sana bila kulazimika kuandika upya msimbo kuwa async.

import threading, time

def download(name):
    time.sleep(1)               # kusubiri mtandao
    print(name, "tayari")

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()                    # yote kwa takriban sekunde 1

Kumbukumbu ya pamoja ni faida na chanzo cha makosa: ikiwa threads mbili zinabadilisha kigezo kimoja, unahitaji threading.Lock. Mara nyingi ni rahisi zaidi kutoshughulikia threads moja kwa moja, bali kutumia pool.

ThreadPoolExecutor

concurrent.futures.ThreadPoolExecutor huunda pool ya threads na kugawa kazi kwao. map inarudisha matokeo kwa mpangilio wa awali.

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)           # yote kwa takriban sekunde 1

GIL ni nini, na kwa nini threads hazifanyi hesabu kwa kasi

GIL (Global Interpreter Lock) ni kufuli ya kimataifa ya mkalimani wa CPython. Kwa kila wakati ni thread moja tu inayotekeleza bytecode ya Python. Thread inapokuwa inasubiri ingizo-pato, huachilia GIL, kwa hiyo kwa mtandao threads zina manufaa. Lakini kwenye hesabu safi, threads nne hazifanyi kazi haraka kuliko moja: zinapokezana kufuli.

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))   # takriban sawa na mfululizo

Maktaba zilizoandikwa kwa C (NumPy na zinazofanana) mara nyingi huachilia GIL wakati wa operesheni nzito, kwa hiyo kwa hizo threads zinaleta kasi.

multiprocessing: sambamba halisi

Kila mchakato una mkalimani wake na GIL yake, kwa hiyo michakato inahesabu kwenye cores tofauti kwa wakati mmoja. Gharama yake ni kumbukumbu tofauti: data kati ya michakato inapitishwa kwa serialization (pickle), na kuanzisha mchakato ni ghali zaidi kuliko thread.

from concurrent.futures import ProcessPoolExecutor

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

if __name__ == "__main__":      # ni lazima kwenye Windows na macOS
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))

Bila ukaguzi wa if __name__ == "__main__", michakato ya mtoto wakati wa import ya moduli itaanza kuunda michakato mipya. Pia kuna moduli ya chini zaidi multiprocessing yenye Process, Pool, Queue, lakini ProcessPoolExecutor ni rahisi zaidi na ina kiolesura sawa na pool ya threads.

Nini cha kuchagua: asyncio, threading au multiprocessing

Python bila GIL na subinterpreters: mambo mapya

Tangu Python 3.13 kuna toleo tofauti bila GIL (free-threading, PEP 703) — linatambulishwa kwa herufi t kwenye jina lake, kwa mfano python3.15t. Katika 3.13 lilikuwa la majaribio, kuanzia 3.14 (PEP 779) linaungwa mkono rasmi. Humo threads zinahesabu kwa sambamba kwenye cores tofauti. Lakini kwa chaguo-msingi bado unapata Python ya kawaida yenye GIL: msimbo wa thread moja kwenye toleo la free-threaded unaendeshwa kidogo polepole, na maktaba zote bado hazijawa tayari kwa hilo.

import sys
print(sys._is_gil_enabled())   # False kwenye toleo la free-threaded

Njia ya pili ya hesabu sambamba ilikuja katika 3.14 (PEP 734) — subinterpreters kwenye moduli concurrent.interpreters: interpreters kadhaa zilizotengwa ndani ya mchakato mmoja, kila moja na GIL yake. Zinatumika kwa urahisi zaidi kupitia InterpreterPoolExecutor, inayoonekana kama pool ya threads. Katika kipimo cha kozi, kazi nne nzito zilichukua 0.39 s dhidi ya 1.03 s kwa thread moja.

Hitimisho

Asyncio python ni zana ya kazi za kusubiri: coroutines zenye async def, pumziko kwa await, na kuanzisha kwa pamoja kupitia TaskGroup au gather. Threads hutatua kazi hiyo hiyo bila kuandika upya msimbo kuwa async, na kwa hesabu nzito zinahitajika michakato, kwa sababu GIL inazuia threads za CPython ya kawaida kuhesabu kwa sambamba. Free-threading na subinterpreters zinabadilisha picha polepole, lakini kanuni ya msingi "tunasubiri — asyncio au threads, tunahesabu — michakato" bado inafanya kazi. Kozi inafafanua haya kwa vitendo: kifuatiliaji cha matumizi kinapata viwango vya sarafu kwa sambamba ndani ya sekunde moja badala ya tatu.