2026年のPython並行処理:AsyncIO、スレッド、プロセス

import asyncio
import httpx

async def fetch_endpoint(client: httpx.AsyncClient, url: str) -> int:
    response = await client.get(url)
    return response.status_code

async def main():
    urls = ["https://httpbin.org/delay/1"] * 20
    async with httpx.AsyncClient(timeout=10) as client:
        tasks = [fetch_endpoint(client, url) for url in urls]
        # Executes all 20 HTTP calls concurrently in ~1.1 seconds on 1 thread!
        results = await asyncio.gather(*tasks)
        print(f"Fetched {len(results)} endpoints successfully.")

asyncio.run(main())

from concurrent.futures import ThreadPoolExecutor
import urllib.request

def blocking_fetch(url: str) -> int:
    # GIL is automatically released during OS socket read
    with urllib.request.urlopen(url, timeout=10) as resp:
        return resp.status

urls = ["https://httpbin.org/delay/1"] * 20
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(blocking_fetch, urls))
    print(f"Threaded fetch completed for {len(results)} URLs.")

from concurrent.futures import ProcessPoolExecutor
import math

def cpu_heavy_hash(n: int) -> int:
    # True parallel execution across separate CPU cores without GIL bottleneck
    return sum(math.isqrt(i) for i in range(n))

if __name__ == "__main__":
    numbers = [50_000_000] * 8
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(cpu_heavy_hash, numbers))
        print(f"Parallel CPU computation finished on {len(results)} cores!")