0Pricing
Python Academy · Lezione

Schemi di I/O asincrono in produzione

Applichi asyncio a client HTTP, code e pipeline reali.

Schemi di I/O asincrono in produzione è una lezione Python Academy gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento Python Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso Python Academy include 4 lezioni in totale.

Buone pratiche per l'event loop

Utilizzi sempre asyncio.run() come unico punto di ingresso. Eviti di mescolare codice sincrono e asincrono senza attenzione e non blocchi mai l'event loop con elaborazioni CPU prolungate.

import asyncio

async def main():
    # Never: time.sleep() — blocks the event loop
    # Always: await asyncio.sleep()
    await asyncio.sleep(1)

asyncio.run(main())

Pooling delle connessioni

Riutilizzi le connessioni invece di aprirne una nuova per ogni richiesta. Librerie come asyncpg e aiohttp forniscono pool integrati.

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@host/db",
        min_size=5, max_size=20
    )
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT 1")
    await pool.close()

Produttore-consumatore con asyncio.Queue

asyncio.Queue disaccoppia i produttori e i consumatori in esecuzione come task concorrenti.

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        await asyncio.sleep(0.5)
    await q.put(None)  # sentinel

async def consumer(q):
    while (item := await q.get()) is not None:
        print(f"processing {item}")
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

Limitazione della frequenza con un semaforo

Limiti le richieste in uscita simultanee per evitare di sovraccaricare le API esterne o di raggiungere i limiti di frequenza.

import asyncio, aiohttp

SEM = asyncio.Semaphore(10)  # 10 concurrent requests max

async def fetch(session, url):
    async with SEM:
        async with session.get(url) as r:
            return await r.json()

async def main(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*[fetch(s, u) for u in urls])

Nuovi tentativi con back-off esponenziale

Racchiuda le operazioni ripetibili in una logica di back-off per gestire con efficacia gli errori temporanei.

import asyncio

async def retry(coro_fn, retries=3, base=1):
    for attempt in range(retries):
        try:
            return await coro_fn()
        except Exception as e:
            if attempt == retries - 1: raise
            delay = base * 2 ** attempt
            await asyncio.sleep(delay)

Arresto controllato

Gestisca asyncio.CancelledError nei task di lunga durata e gestisca i segnali del sistema operativo per arrestare l'applicazione in modo ordinato.

import asyncio, signal

async def server():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, stop.set)
    await stop.wait()
    print("Shutting down gracefully")

Risposte in streaming

Utilizzi i generatori asincroni per trasmettere risposte di grandi dimensioni senza caricarle interamente in memoria.

import asyncio, aiohttp

async def stream(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            async for chunk in r.content.iter_chunked(4096):
                process(chunk)

async def process(chunk):
    await asyncio.sleep(0)
    print(len(chunk), "bytes")

Task in background con FastAPI

BackgroundTasks di FastAPI accoda il lavoro da eseguire dopo l'invio della risposta, mantenendo rapidi gli endpoint.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, body: str):
    # heavy synchronous work
    pass

@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email, "Welcome!")
    return {"status": "registered"}

Pattern heartbeat

Esegua un task in background periodico insieme al ciclo principale del server utilizzando create_task.

import asyncio

async def heartbeat():
    while True:
        print("alive")
        await asyncio.sleep(30)

async def main():
    asyncio.create_task(heartbeat())
    await serve_forever()

Profilazione del codice asincrono

Utilizzi yappi o py-spy per la profilazione compatibile con il codice asincrono. Abiliti la modalità di debug di asyncio per ricevere avvisi sui callback lenti.

# Enable debug to catch slow callbacks:
# asyncio.run(main(), debug=True)

# PYTHONASYNCIODEBUG=1 python server.py
# logs: Executing <Task coroutine=...> took 0.150 seconds

Come evitare le insidie più comuni

Regole fondamentali: non chiami mai codice bloccante nell'event loop, attenda sempre le coroutine, annulli i task durante l'arresto, utilizzi pool di connessioni e limiti la concorrenza con i semafori.

import asyncio

async def bad():
    import time
    time.sleep(2)        # blocks the entire event loop!

async def good():
    await asyncio.sleep(2)  # yields control

Verifica rapida

Quale primitiva di asyncio disaccoppia i produttori e i consumatori in esecuzione come task concorrenti?

Riepilogo

Pattern asincroni per la produzione: utilizzi pool di connessioni, controlli la concorrenza con Semaphore, implementi tentativi ripetuti con back-off, trasmetta i dati di grandi dimensioni in streaming, esegua un arresto controllato in risposta ai segnali ed esegua task periodici con create_task.

Domande Frequenti

La lezione «Schemi di I/O asincrono in produzione» è gratuita?

Sì — il testo completo di «Schemi di I/O asincrono in produzione» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso Python Academy, passa a CoddyKit PRO. Il corso Python Academy include 4 lezioni in totale.

Cosa imparerò in «Schemi di I/O asincrono in produzione»?

Applichi asyncio a client HTTP, code e pipeline reali. Eserciti Python Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare Python Academy?

Non è richiesta alcuna esperienza precedente. Python Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.

Quanto tempo richiede la lezione «Schemi di I/O asincrono in produzione»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione Python Academy?

Sì. Ogni lezione Python Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Coroutine e ciclo degli eventi
  2. await, task e raccolta dei risultati
  3. Context manager e iteratori asincroni
  4. Schemi di I/O asincrono in produzione
← Torna a Python Academy