파이썬 비동기 프로그래밍의 진화

파이썬에서 동시성을 구현하는 방식은 여러 단계를 거쳐 발전했습니다. 초기에는 greenlet 기반의 gevent를 통해 수동 전환 방식을 사용했고, yield 키워드로 코루틴을 흉내 내기도 했습니다. 파이썬 3.4에서 asyncio가 표준 라이브러리로 추가되었고, 3.5에서는 async/await 구문이 도입되어 비동기 코드의 가독성이 크게 향상되었습니다.

이벤트 루프의 핵심 메커니즘

asyncio의 핵심은 이벤트 루프입니다. 이 루프는 다수의 작업을 관리하며, 실행 가능한 작업과 완료된 작업을 구분하여 효율적으로 스케줄링합니다.

# 이벤트 루프 획득 및 작업 실행 (3.6 이하)
ev_loop = asyncio.get_event_loop()
ev_loop.run_until_complete(asyncio.wait(jobs))

# 파이썬 3.7+ 간소화된 방식
asyncio.run(main_coroutine())

코루틴 함수와 객체의 구분

async def로 정의된 함수는 코루틴 함수이며, 이를 호출하면 코루틴 객체가 반환됩니다. 중요한 점은 이 객체 자체를 호출해도 내부 코드가 실행되지 않으며, 반드시 이벤트 루프에 등록해야 실행됩니다.

import asyncio

async def worker(name):
    print(f"[{name}] 작업 시작")
    await asyncio.sleep(1)
    return f"[{name}] 완료"

# 코루틴 객체 생성 (실행되지 않음)
coro_obj = worker("Alpha")

# 이벤트 루프에 등록하여 실행
asyncio.run(coro_obj)

await와 비동기 대기

await 키워드는 코루틴, Future, Task 등 대기 가능 객체(Awaitable) 뒤에 사용됩니다. IO 작업이 완료될 때까지 현재 코루틴의 실행을 일시 중지하며, 이때 제어권은 이벤트 루프에 반환되어 다른 작업이 실행될 수 있습니다.

import asyncio

async def fetch_data(source):
    print(f"{source}: 데이터 요청 중...")
    await asyncio.sleep(2)  # IO 작업 시뮬레이션
    print(f"{source}: 응답 수신")
    return f"데이터-{source}"

async def orchestrator():
    # 순차 실행: 각 작업이 완료될 때까지 대기
    first = await fetch_data("API-A")
    print(f"첫 번째 결과: {first}")
    
    second = await fetch_data("API-B")
    print(f"두 번째 결과: {second}")

asyncio.run(orchestrator())

Task: 병렬 실행의 핵심

Task는 코루틴을 이벤트 루프에 등록하여 동시에 실행할 수 있게 만드는 객체입니다. asyncio.create_task()(3.7+) 또는 asyncio.ensure_future()로 생성합니다.

import asyncio

async def compute(value):
    print(f"계산 시작: {value}")
    await asyncio.sleep(1)
    return value ** 2

async def parallel_main():
    # 동시에 여러 Task 생성
    t1 = asyncio.create_task(compute(5))
    t2 = asyncio.create_task(compute(8))
    t3 = asyncio.ensure_future(compute(12))
    
    # 모든 결과 수집
    r1, r2, r3 = await t1, await t2, await t3
    print(f"결과: {r1}, {r2}, {r3}")

asyncio.run(parallel_main())

대량의 Task를 효율적으로 관리하려면 asyncio.wait()를 활용합니다. 이 함수는 완료된 작업과 대기 중인 작업을 분리하여 반환합니다.

async def batch_processor():
    tasks = [compute(i) for i in range(10)]
    
    # timeout 파라미터로 최대 대기 시간 설정 가능
    completed, pending = await asyncio.wait(
        tasks, 
        timeout=5.0,
        return_when=asyncio.ALL_COMPLETED
    )
    
    for done_task in completed:
        print(f"완료: {done_task.result()}")

asyncio.run(batch_processor())

Future: 저수준 비동기 결과 저장소

Future는 비동기 작업의 결과를 담는 저수준 컨테이너입니다. 일반적으로 직접 사용하지 않고 Task를 통해 간접적으로 활용하지만, 수동 결과 설정이나 취소 등 세밀한 제어가 필요할 때 유용합니다.

import asyncio

async def manual_future_demo():
    loop = asyncio.get_event_loop()
    fut = loop.create_future()
    
    # 비동기적으로 결과 설정
    asyncio.create_task(set_result_later(fut, "수동 결과", 1))
    
    result = await fut
    print(f"Future 결과: {result}")

async def set_result_later(future, value, delay):
    await asyncio.sleep(delay)
    future.set_result(value)

asyncio.run(manual_future_demo())

asyncio.Future와 concurrent.futures.Future는 다른 용도로 설계되었습니다. 전자는 코루틴 기반 비동기 환경에서 await로 결과를 기다리고, 후자는 스레드/프로세스 풀에서 .result()로 동기적으로 결과를 얻습니다.

동기 함수를 비동기 환경에서 실행하기

블로킹 동기 함수(예: 데이터베이스 드라이버)를 코루틴에서 호출해야 할 때 run_in_executor()를 사용합니다. 이 메서드는 스레드 풀이나 프로세스 풀에서 동기 코드를 실행하여 이벤트 루프를 블록하지 않습니다.

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

def blocking_io(identifier):
    print(f"블로킹 작업 {identifier} 시작")
    time.sleep(2)  # 동기 블로킹 호출
    return identifier * 10

async def async_wrapper(pool, idx):
    loop = asyncio.get_event_loop()
    # executor에 None을 전달하면 기본 스레드 풀 사용
    # 특정 풀을 전달하면 해당 풀에서 실행
    result = await loop.run_in_executor(pool, blocking_io, idx)
    print(f"변환 결과: {result}")

# 스레드 풀 크기 제한으로 동시 실행 수 제어
custom_pool = ThreadPoolExecutor(max_workers=4)
task_list = [async_wrapper(custom_pool, i) for i in range(20)]

asyncio.run(asyncio.wait(task_list))

고급 병렬 처리: gather vs as_completed

asyncio.gather: 순서 보장 결과 수집

모든 작업이 완료될 때까지 대기하며, 입력 순서와 동일한 순서로 결과를 반환합니다.

async def gather_demo():
    async def delayed_echo(msg, wait):
        await asyncio.sleep(wait)
        return f"{msg} (대기 {wait}초)"
    
    results = await asyncio.gather(
        delayed_echo("첫째", 3),
        delayed_echo("둘째", 1),
        delayed_echo("셋째", 2)
    )
    # 출력: ['첫째 (대기 3초)', '둘째 (대기 1초)', '셋째 (대기 2초)']
    print(results)

asyncio.run(gather_demo())

asyncio.as_completed: 완료 순서 결과 처리

작업이 완료되는 순서대로 결과를 반환하는 이터레이터를 제공합니다. 빠른 응답을 먼저 처리해야 할 때 유용합니다.

async def as_completed_demo():
    async def variable_delay(n):
        await asyncio.sleep(n)
        return f"작업-{n} 완료"
    
    pending = [variable_delay(i) for i in [3, 1, 4, 1, 5]]
    
    for completed in asyncio.as_completed(pending):
        result = await completed
        print(result)  # 1초 작업이 먼저 출력됨

asyncio.run(as_completed_demo())

성능 향상: uvloop 적용

uvloop는 asyncio의 이벤트 루프를 Cython으로 재구현한 고성능 대체품입니다. 일반적으로 2배 이상의 성능 향상을 제공합니다.

import asyncio
import uvloop

# uvloop를 기본 이벤트 루프 정책으로 설정
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# 이후 모든 asyncio 코드는 uvloop 기반으로 실행
asyncio.run(high_performance_app())

주의: 파이썬 3.7 이상에서만 지원됩니다.

콜백 패턴 구현

동기 콜백

def sync_callback(future_task):
    print(f"콜백 수신: {future_task.result()}")

async def task_with_callback():
    job = asyncio.create_task(worker("콜백 테스트"))
    job.add_done_callback(sync_callback)
    await job

asyncio.run(task_with_callback())

비동기 콜백

콜백 함수 자체도 비동기 처리가 필요한 경우, 별도의 코루틴으로 연결합니다.

async def async_callback(previous_task):
    await asyncio.sleep(0.5)  # 추가 비동기 작업
    print(f"비동기 콜백: {previous_task.result()}")

async def chain_async_operations():
    first = asyncio.create_task(worker("연쇄 작업"))
    # 첫 작업 완료 후 콜백 실행
    await first
    await async_callback(first)

asyncio.run(chain_async_operations())

동시성 제어: Semaphore

리소스 사용량이나 외부 API 호출 빈도를 제한해야 할 때 Semaphore를 사용합니다.

import asyncio

# 최대 3개 동시 실행 제한
semaphore = asyncio.Semaphore(3)

async def rate_limited_task(task_id):
    async with semaphore:
        print(f"작업 {task_id} 실행 중")
        await asyncio.sleep(2)
        print(f"작업 {task_id} 완료")

async def controlled_execution():
    await asyncio.gather(*[rate_limited_task(i) for i in range(10)])

asyncio.run(controlled_execution())

주의사항: 파이썬 3.7+에서 asyncio.run() 사용 시, Semaphore는 내부에서 생성된 이벤트 루프와 동일한 스코프에서 초기화해야 합니다. 그렇지 않으면 RuntimeError: ... attached to a different loop 오류가 발생합니다.

# 올바른 사용법 (3.7+)
async def main():
    sem = asyncio.Semaphore(5)  # main 내부에서 생성
    await asyncio.gather(*[limited_work(sem, i) for i in range(20)])

asyncio.run(main())

다중 이벤트 루프 관리

특수한 경우 여러 이벤트 루프를 병렬로 운영할 수 있습니다.

import asyncio

async def loop_specific_task(name, delay):
    await asyncio.sleep(delay)
    return f"{name} 결과"

# 기본 루프
primary_loop = asyncio.get_event_loop()

# 새로운 루프 생성 및 설정
secondary_loop = asyncio.new_event_loop()
asyncio.set_event_loop(secondary_loop)

print(primary_loop is secondary_loop)  # False

# 각 루프에서 독립적으로 실행
task_a = secondary_loop.create_task(loop_specific_task("A", 1))
task_b = primary_loop.create_task(loop_specific_task("B", 2))

secondary_loop.run_until_complete(asyncio.wait([task_a]))
primary_loop.run_until_complete(asyncio.wait([task_b]))

실전 예제: 비동기 웹 크롤러

다음은 asyncio와 aiomysql, httpx를 활용한 완전한 비동기 크롤러 예제입니다.

import asyncio
import aiomysql
import httpx
from lxml import etree
from urllib.parse import urljoin
import re

class AsyncCrawler:
    def __init__(self, start_url, db_config):
        self.start_url = start_url
        self.db_config = db_config
        self.visited = set()
        self.pending = [start_url]
        self.active = True
    
    async def fetch_page(self, url, client):
        try:
            response = await client.get(url)
            if response.status_code == 200:
                return response
        except Exception as exc:
            print(f"요청 실패: {url} - {exc}")
        return None
    
    def extract_links(self, base_url, html_content):
        tree = etree.HTML(html_content)
        links = tree.xpath(".//a/@href")
        
        valid_links = []
        for link in links:
            if re.match(r"^/.*", link):
                absolute = urljoin(base_url, link)
                if absolute not in self.visited:
                    valid_links.append(absolute)
        return valid_links
    
    async def parse_and_store(self, pool, response):
        tree = etree.HTML(response.text)
        titles = tree.xpath(".//h3[@class='info-title']//text()")
        
        if not titles:
            return
        
        print(f"발견: {len(titles)}개 제목 from {response.url}")
        
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                records = [(t,) for t in titles]
                await cur.executemany(
                    "INSERT INTO articles (title) VALUES (%s)",
                    records
                )
    
    async def consumer(self, pool):
        async with httpx.AsyncClient() as client:
            while self.active:
                if not self.pending:
                    await asyncio.sleep(3)
                    continue
                
                current_url = self.pending.pop(0)
                if current_url in self.visited:
                    continue
                
                self.visited.add(current_url)
                print(f"크롤링: {current_url}")
                
                resp = await self.fetch_page(current_url, client)
                if resp and len(resp.text) > 100:
                    new_links = self.extract_links(self.start_url, resp.text)
                    self.pending.extend(new_links)
                    await self.parse_and_store(pool, resp)
    
    async def initialize(self):
        async with httpx.AsyncClient() as client:
            resp = await self.fetch_page(self.start_url, client)
            if resp:
                initial_links = self.extract_links(self.start_url, resp.text)
                self.pending.extend(initial_links)
    
    async def run(self):
        loop = asyncio.get_event_loop()
        
        # 데이터베이스 연결 풀 생성
        pool = await aiomysql.create_pool(
            **self.db_config,
            loop=loop,
            autocommit=True
        )
        
        try:
            await self.initialize()
            await self.consumer(pool)
        finally:
            pool.close()
            await pool.wait_closed()

# 실행
config = {
    'host': '127.0.0.1',
    'port': 3306,
    'user': 'root',
    'password': 'secret',
    'db': 'crawler_db'
}

crawler = AsyncCrawler("https://example.com/blog", config)
asyncio.run(crawler.run())

핵심 메서드 비교 정리

메서드	용도	특징
asyncio.wait()	다중 작업 상태 관리	완료/대기 작업 분리 반환, 조건 설정 가능
asyncio.wait_for()	단일 작업 타임아웃	시간 초과 시 TimeoutError 발생
asyncio.gather()	결과 순서 보장 수집	모든 결과를 입력 순서대로 리스트로 반환
asyncio.as_completed()	완료 순서 처리	이터레이터로 빠른 완료 작업 먼저 반환

적절한 도구 선택은 애플리케이션의 요구사항에 따라 달라집니다: 전체 결과가 필요하면 gather, 빠른 응답 우선 처리는 as_completed, 세밀한 상태 제어는 wait를 활용하세요.