Python 비동기 프로그래밍(AsyncIO) 개념 정리

1️⃣ AsyncIO 기초

AsyncIO는 파이썬의 비동기 프로그래밍을 위한 라이브러리이다.

import asyncio

async def hello():
    print('Hello')
    await asyncio.sleep(1)
    print('World')

# 실행
asyncio.run(hello())

✅ 특징:

비동기 프로그래밍 지원
코루틴 기반 동작
단일 스레드에서 동시성 구현

2️⃣ 코루틴과 태스크

코루틴은 일시 중단되고 나중에 재개될 수 있는 비동기 함수이며, 태스크는 코루틴을 실행하는 단위이다.

async def task1():
    print('Task 1 시작')
    await asyncio.sleep(2)
    print('Task 1 종료')
    return 'Task 1 결과'

async def task2():
    print('Task 2 시작')
    await asyncio.sleep(1)
    print('Task 2 종료')
    return 'Task 2 결과'

async def main():
    # 동시에 여러 태스크 실행
    results = await asyncio.gather(
        task1(),
        task2()
    )
    print(f'결과: {results}')

asyncio.run(main())

✅ 특징:

여러 태스크 동시 실행
결과 수집 가능
비동기 작업 조율

3️⃣ 이벤트 루프

이벤트 루프는 AsyncIO의 핵심 실행 장치로, 비동기 태스크를 관리하고 실행한다.

async def long_operation():
    print('시작')
    await asyncio.sleep(3)
    print('종료')
    return '완료'

async def main():
    # 이벤트 루프 가져오기
    loop = asyncio.get_running_loop()
    
    # 태스크 생성
    task = loop.create_task(long_operation())
    
    # 다른 작업 수행
    print('다른 작업 수행 중...')
    
    # 태스크 완료 대기
    result = await task
    print(f'결과: {result}')

# 실행
asyncio.run(main())

✅ 특징:

이벤트 루프 직접 제어
태스크 생성 및 관리
비동기 작업 스케줄링

4️⃣ 비동기 컨텍스트 매니저

비동기 컨텍스트 매니저는 비동기 리소스를 안전하게 관리하는 데 사용된다.

class AsyncResource:
    async def __aenter__(self):
        print('리소스 획득')
        await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print('리소스 해제')
        await asyncio.sleep(1)

async def use_resource():
    async with AsyncResource() as resource:
        print('리소스 사용 중')
        await asyncio.sleep(1)

# 실행
asyncio.run(use_resource())

✅ 특징:

비동기 리소스 관리
자동 정리 보장
컨텍스트 기반 작업

5️⃣ 비동기 반복자

비동기 반복자는 데이터를 비동기적으로 순회할 수 있게 해준다.

class AsyncCounter:
    def __init__(self, limit):
        self.limit = limit
        self.counter = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.counter < self.limit:
            self.counter += 1
            await asyncio.sleep(0.1)
            return self.counter
        raise StopAsyncIteration

async def use_counter():
    async for i in AsyncCounter(5):
        print(f'카운트: {i}')

# 실행
asyncio.run(use_counter())

✅ 특징:

비동기 반복 지원
지연 실행 가능
메모리 효율적 처리

6️⃣ 실용적인 예제

실제 애플리케이션에서 AsyncIO를 활용하는 다양한 예제를 살펴보자.

비동기 웹 요청:

import aiohttp
import asyncio

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        'http://example.com',
        'http://example.org',
        'http://example.net'
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for url, html in zip(urls, results):
            print(f'{url}: {len(html)} bytes')

asyncio.run(main())

비동기 데이터베이스 작업:

import asyncpg

async def create_table(pool):
    async with pool.acquire() as conn:
        await conn.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id SERIAL PRIMARY KEY,
                name TEXT,
                email TEXT
            )
        ''')

async def insert_user(pool, name, email):
    async with pool.acquire() as conn:
        return await conn.fetchval(
            'INSERT INTO users(name, email) VALUES($1, $2) RETURNING id',
            name, email
        )

async def main():
    # 데이터베이스 연결 풀 생성
    pool = await asyncpg.create_pool(
        user='user',
        password='password',
        database='dbname',
        host='localhost'
    )
    
    await create_table(pool)
    
    # 비동기로 여러 사용자 추가
    users = [
        ('User1', '[email protected]'),
        ('User2', '[email protected]'),
        ('User3', '[email protected]')
    ]
    
    tasks = [insert_user(pool, name, email) for name, email in users]
    user_ids = await asyncio.gather(*tasks)
    print(f'추가된 사용자 ID: {user_ids}')
    
    await pool.close()

비동기 파일 처리:

import aiofiles
import asyncio

async def read_large_file(filename):
    async with aiofiles.open(filename, mode='r') as f:
        contents = await f.read()
    return contents

async def write_file(filename, data):
    async with aiofiles.open(filename, mode='w') as f:
        await f.write(data)

async def process_files():
    # 동시에 여러 파일 읽기
    files = ['file1.txt', 'file2.txt', 'file3.txt']
    tasks = [read_large_file(file) for file in files]
    
    contents = await asyncio.gather(*tasks)
    
    # 처리된 데이터 저장
    processed = [f"Processed: {content[:100]}..." for content in contents]
    
    # 동시에 여러 파일 쓰기
    write_tasks = [
        write_file(f"processed_{file}", data) 
        for file, data in zip(files, processed)
    ]
    
    await asyncio.gather(*write_tasks)
    print("모든 파일 처리 완료")

# 실행
asyncio.run(process_files())

비동기 큐 처리:

import asyncio
import random

async def producer(queue, id):
    for i in range(5):
        # 작업 생성
        item = f"생산자 {id} - 항목 {i}"
        
        # 큐에 항목 추가
        await queue.put(item)
        print(f"생산됨: {item}")
        
        # 랜덤 지연
        await asyncio.sleep(random.uniform(0.1, 0.5))

async def consumer(queue, id):
    while True:
        # 큐에서 항목 가져오기
        item = await queue.get()
        
        # 항목 처리
        print(f"소비자 {id} 처리 중: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.6))
        
        # 작업 완료 표시
        queue.task_done()

async def main():
    # 큐 생성
    queue = asyncio.Queue()
    
    # 생산자 및 소비자 태스크 생성
    producers = [producer(queue, i) for i in range(3)]
    consumers = [asyncio.create_task(consumer(queue, i)) for i in range(2)]
    
    # 생산자 실행 및 대기
    await asyncio.gather(*producers)
    
    # 모든 항목이 처리될 때까지 대기
    await queue.join()
    
    # 소비자 태스크 취소
    for c in consumers:
        c.cancel()

# 실행
asyncio.run(main())

✅ 주요 팁:

I/O 바운드 작업에 적합
동기 코드와 혼합 시 주의
이벤트 루프 블로킹 피하기
적절한 에러 처리 구현
비동기 컨텍스트 매니저 활용
태스크 취소 처리 구현
디버깅 도구 활용
효율적인 태스크 그룹화
코루틴 체인 구성 시 예외 전파 고려
백그라운드 태스크 관리

7️⃣ 동시성 제어 도구

AsyncIO는 다양한 동시성 제어 도구를 제공하여 복잡한 비동기 작업을 관리한다.

세마포어:

import asyncio

async def worker(semaphore, id):
    async with semaphore:
        # 세마포어로 제한된 구역
        print(f"작업자 {id} 시작")
        await asyncio.sleep(1)
        print(f"작업자 {id} 종료")

async def main():
    # 최대 3개의 동시 태스크만 허용
    semaphore = asyncio.Semaphore(3)
    
    # 10개의 작업 동시 실행 (최대 3개씩 처리)
    tasks = [worker(semaphore, i) for i in range(10)]
    await asyncio.gather(*tasks)

# 실행
asyncio.run(main())

락(Lock):

import asyncio

async def protected_resource(lock, id):
    # 락 획득
    async with lock:
        print(f"리소스 사용 중 (작업자 {id})")
        await asyncio.sleep(1)
        print(f"리소스 사용 완료 (작업자 {id})")

async def main():
    # 락 생성
    lock = asyncio.Lock()
    
    # 여러 태스크가 동일한 락을 공유
    tasks = [protected_resource(lock, i) for i in range(5)]
    await asyncio.gather(*tasks)

# 실행
asyncio.run(main())

이벤트:

import asyncio

async def waiter(event, id):
    print(f"대기자 {id}: 이벤트 대기 중")
    await event.wait()
    print(f"대기자 {id}: 이벤트 발생! 작업 진행")

async def setter(event):
    print("준비 중...")
    await asyncio.sleep(2)
    print("이벤트 설정!")
    
    # 모든 대기 중인 태스크에 신호 보내기
    event.set()

async def main():
    # 이벤트 생성
    event = asyncio.Event()
    
    # 대기 태스크와 설정 태스크 생성
    waiters = [asyncio.create_task(waiter(event, i)) for i in range(5)]
    setter_task = asyncio.create_task(setter(event))
    
    # 모든 태스크 완료 대기
    await asyncio.gather(setter_task, *waiters)

# 실행
asyncio.run(main())

조건(Condition):

import asyncio
import random

async def consumer(condition, id, buffer):
    while True:
        # 조건 획득 및 대기
        async with condition:
            # 버퍼가 비어있으면 대기
            while len(buffer) == 0:
                print(f"소비자 {id}: 버퍼가 비어있어 대기 중")
                await condition.wait()
            
            # 항목 소비
            item = buffer.pop(0)
            print(f"소비자 {id}: {item} 처리")
            
            # 생산자에게 신호
            condition.notify()
        
        await asyncio.sleep(random.uniform(0.1, 0.3))

async def producer(condition, id, buffer, max_size):
    for i in range(5):
        item = f"항목 {i} (생산자 {id})"
        
        # 조건 획득
        async with condition:
            # 버퍼가 가득 찼으면 대기
            while len(buffer) >= max_size:
                print(f"생산자 {id}: 버퍼가 가득 차서 대기 중")
                await condition.wait()
            
            # 항목 생산
            buffer.append(item)
            print(f"생산자 {id}: {item} 생성")
            
            # 소비자에게 신호
            condition.notify()
        
        await asyncio.sleep(random.uniform(0.1, 0.5))

async def main():
    # 공유 버퍼와 조건 변수
    buffer = []
    max_size = 2
    condition = asyncio.Condition()
    
    # 생산자 및 소비자 태스크
    producers = [producer(condition, i, buffer, max_size) for i in range(2)]
    consumers = [asyncio.create_task(consumer(condition, i, buffer)) for i in range(2)]
    
    # 생산자 완료 대기
    await asyncio.gather(*producers)
    
    # 소비자 태스크 취소
    for c in consumers:
        c.cancel()
    
    # 취소된 태스크 예외 처리
    await asyncio.gather(*consumers, return_exceptions=True)

# 실행
asyncio.run(main())

✅ 특징:

멀티태스킹 조정
리소스 경쟁 관리
태스크 간 동기화
비동기 작업 흐름 제어
효율적인 병렬 처리

8️⃣ 비동기 스트림 처리

대용량 데이터를 메모리 효율적으로 처리하기 위한 비동기 스트림 기법이다.

import asyncio

async def number_generator(limit):
    # 스트림 소스
    for i in range(limit):
        await asyncio.sleep(0.1)  # 데이터 생성 시간 흉내
        yield i

async def process_numbers():
    # 비동기 생성기에서 데이터 스트리밍
    async for number in number_generator(10):
        result = number * number
        print(f"처리: {number} -> {result}")

# 실행
asyncio.run(process_numbers())

스트림 파이프라인:

import asyncio

async def data_source():
    # 데이터 소스 (API, 파일 등)
    for i in range(10):
        await asyncio.sleep(0.2)
        yield i

async def filter_even(stream):
    # 짝수만 필터링
    async for item in stream:
        if item % 2 == 0:
            yield item

async def multiply(stream, factor=2):
    # 각 항목에 factor 곱하기
    async for item in stream:
        yield item * factor

async def process_pipeline():
    # 파이프라인 구성
    source = data_source()
    filtered = filter_even(source)
    multiplied = multiply(filtered, 3)
    
    # 결과 소비
    async for result in multiplied:
        print(f"결과: {result}")

# 실행
asyncio.run(process_pipeline())

✅ 특징:

메모리 효율적인 데이터 처리
대용량 데이터셋 처리에 적합
비동기 파이프라인 구성
백프레셔 관리
스트림 변환 및 필터링

주요 팁

✅ AsyncIO 모범 사례:

CPU 바운드 작업 분리: CPU 집약적 작업은 concurrent.futures.ProcessPoolExecutor로 처리
타임아웃 설정: 모든 외부 호출에 타임아웃 적용하여 무한 대기 방지
디버깅 활성화: asyncio.run(main(), debug=True)로 디버깅 강화
태스크 그룹화: 관련 태스크는 asyncio.TaskGroup 또는 asyncio.gather로 그룹화
적절한 취소 처리: 태스크 취소 시 try/except asyncio.CancelledError 패턴 사용
프로메테우스 방지: 메인 이벤트 루프를 블로킹하는 코드 피하기
백오프 재시도: 외부 서비스 호출 시 지수 백오프 적용
컨텍스트 관리자 활용: 리소스 관리에 async with 사용
구조화된 동시성: 부모 태스크가 자식 태스크의 생명주기 관리
비동기 작업 취소: 더 이상 필요 없는 태스크는 명시적으로 취소
효율적인 I/O 처리: 파일 I/O는 aiofiles 사용, 네트워크는 aiohttp 사용
예외 처리: 모든 비동기 코드에 적절한 예외 처리 구현
코드 가독성 유지: 복잡한 비동기 로직은 함수로 분리
이벤트 루프 커스터마이징: 필요시 이벤트 루프 정책 커스터마이징
로깅 통합: 비동기 코드에도 일관된 로깅 전략 적용

KR_AsyncIO - somaz94/python-study GitHub Wiki

Python 비동기 프로그래밍(AsyncIO) 개념 정리

1️⃣ AsyncIO 기초

2️⃣ 코루틴과 태스크

3️⃣ 이벤트 루프

4️⃣ 비동기 컨텍스트 매니저

5️⃣ 비동기 반복자

6️⃣ 실용적인 예제

비동기 웹 요청:

비동기 데이터베이스 작업:

비동기 파일 처리:

비동기 큐 처리:

7️⃣ 동시성 제어 도구

세마포어:

락(Lock):

이벤트:

조건(Condition):

8️⃣ 비동기 스트림 처리

스트림 파이프라인:

주요 팁

⚠️ GitHub.com Fallback ⚠️

KR_AsyncIO - somaz94/python-study GitHub Wiki

Python 비동기 프로그래밍(AsyncIO) 개념 정리

1️⃣ AsyncIO 기초

2️⃣ 코루틴과 태스크

3️⃣ 이벤트 루프

4️⃣ 비동기 컨텍스트 매니저

5️⃣ 비동기 반복자

6️⃣ 실용적인 예제

비동기 웹 요청:

비동기 데이터베이스 작업:

비동기 파일 처리:

비동기 큐 처리:

7️⃣ 동시성 제어 도구

세마포어:

락(Lock):

이벤트:

조건(Condition):

8️⃣ 비동기 스트림 처리

스트림 파이프라인:

주요 팁

⚠️ **GitHub.com Fallback** ⚠️

⚠️ GitHub.com Fallback ⚠️