비동기 패턴 (fan-out/fan-in)

Python code

import asyncio

async def fetch_chunk(chunk_id: int, data: list) -> dict:
    """개별 청크 처리 (fan-out)"""
    await asyncio.sleep(0.3)
    result = sum(data)
    print(f"  청크 {chunk_id}: 합계={result}")
    return {"chunk": chunk_id, "sum": result}

def split_data(data: list, n_chunks: int) -> list[list]:
    """데이터를 n개 청크로 분할"""
    size = len(data) // n_chunks
    return [data[i*size:(i+1)*size] for i in range(n_chunks)]

async def parallel_sum(data: list, n_workers: int = 4) -> int:
    """fan-out/fan-in 패턴"""
    # Fan-out: 작업 분산
    chunks = split_data(data, n_workers)
    tasks = [
        fetch_chunk(i, chunk)
        for i, chunk in enumerate(chunks)
    ]

    # Fan-in: 결과 수집
    results = await asyncio.gather(*tasks)
    total = sum(r["sum"] for r in results)
    return total

async def pipeline_pattern():
    """비동기 파이프라인"""
    q1 = asyncio.Queue()
    q2 = asyncio.Queue()

    async def stage1():
        for i in range(5):
            await q1.put(i * 10)
        await q1.put(None)

    async def stage2():
        while (item := await q1.get()) is not None:
            await q2.put(item + 1)
        await q2.put(None)

    async def stage3():
        results = []
        while (item := await q2.get()) is not None:
            results.append(item * 2)
        print(f"파이프라인 결과: {results}")

    await asyncio.gather(stage1(), stage2(), stage3())

async def main():
    data = list(range(100))
    total = await parallel_sum(data)
    print(f"병렬 합계: {total}")  # 4950

    await pipeline_pattern()

asyncio.run(main())