EP14 | 고급 Python 활용 #3 | 비동기 프로그래밍 (AsyncIO)

이 글에서 다루는 개념

Python의 **비동기 프로그래밍(Asynchronous Programming)**을 활용하면
시간이 오래 걸리는 작업을 효율적으로 실행할 수 있습니다.
이번 글에서는 다음 내용을 학습합니다.

• 동기(Synchronous) vs 비동기(Asynchronous) 차이
• async와 await 키워드 활용
• asyncio 모듈을 사용한 비동기 처리
• 비동기 I/O와 네트워크 요청 처리

---

1️⃣ 동기(Synchronous) vs 비동기(Asynchronous) 차이

📌 동기 방식 (Synchronous)

• 하나의 작업이 완료될 때까지 다음 작업이 실행되지 않음
• 코드가 순차적으로 실행되므로 실행 시간이 오래 걸릴 수 있음

import time

def task(name):
    print(f"{name} 시작")
    time.sleep(2)
    print(f"{name} 완료")

task("작업 1")
task("작업 2")

 

⏳ 출력 (총 4초 소요)

작업 1 시작  
(2초 대기)  
작업 1 완료  
작업 2 시작  
(2초 대기)  
작업 2 완료  

---

📌 비동기 방식 (Asynchronous)

• 여러 작업을 동시에 실행할 수 있음
• 하나의 작업이 대기 상태(await)에 들어가면 다른 작업이 실행됨

import asyncio

async def task(name):
    print(f"{name} 시작")
    await asyncio.sleep(2)
    print(f"{name} 완료")

async def main():
    await asyncio.gather(task("작업 1"), task("작업 2"))

asyncio.run(main())

 

⚡ 출력 (총 2초 소요, 동시에 실행됨!)

작업 1 시작  
작업 2 시작  
(2초 대기)  
작업 1 완료  
작업 2 완료  

---

2️⃣ async와 await 키워드 사용법

🔹 비동기 함수 (async def)

Python에서 비동기 함수는 async def 키워드를 사용하여 정의합니다.

import asyncio

async def my_async_function():
    print("비동기 함수 실행 중...")
    await asyncio.sleep(2)  # 2초 대기 (비동기)
    print("비동기 함수 완료!")

asyncio.run(my_async_function())  # 실행

 

📌 비동기 함수 실행 방법

• async def로 정의된 함수는 일반적으로 await를 사용하여 실행해야 함
• asyncio.run(함수())를 사용하여 실행

---

3️⃣ asyncio.gather()를 사용한 동시에 여러 작업 실행

📌 여러 개의 비동기 작업을 동시에 실행할 때 asyncio.gather()를 사용

import asyncio

async def task(name, delay):
    print(f"{name} 시작")
    await asyncio.sleep(delay)
    print(f"{name} 완료")

async def main():
    await asyncio.gather(
        task("작업 1", 2),
        task("작업 2", 3),
        task("작업 3", 1)
    )

asyncio.run(main())

 

📌 비동기 실행 순서

1. 모든 작업이 동시에 시작됨
2. 작업 3 → 작업 1 → 작업 2 순서로 완료됨 (가장 짧은 작업이 먼저 끝남)

---

4️⃣ 비동기 네트워크 요청 처리 (aiohttp)

Python의 aiohttp 라이브러리를 사용하면 비동기 HTTP 요청을 처리할 수 있습니다.

 

📌 설치 (aiohttp가 없을 경우)

pip install aiohttp

 

📌 비동기 방식으로 여러 웹페이지 요청하기

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.text()
            print(f"{url} 응답 길이: {len(data)}")

async def main():
    urls = ["https://example.com", "https://www.python.org", "https://www.github.com"]
    await asyncio.gather(*(fetch(url) for url in urls))

asyncio.run(main())

 

📌 동작 방식

• aiohttp.ClientSession() → 비동기 HTTP 세션 생성
• session.get(url) → 비동기 GET 요청
• await response.text() → 응답 본문을 비동기적으로 읽음
• 모든 요청을 동시에 실행하여 응답 속도가 빨라짐

---

📌 실전 문제: 비동기 프로그래밍 연습하기

---

✅ 문제 1: async와 await을 사용하여 3초 대기 후 메시지 출력하기

📌 비동기 함수를 사용하여 3초 후 "완료!"를 출력하세요.

import asyncio

async def wait_and_print():
    # 🔽 여기에 코드 작성

import asyncio

async def wait_and_print():
    print("3초 대기 중...")
    await asyncio.sleep(3)
    print("완료!")

asyncio.run(wait_and_print())

---

✅ 문제 2: asyncio.gather()를 사용하여 동시에 3개의 작업 실행하기

📌 각 작업이 다른 시간을 대기한 후 "작업 완료!"를 출력하세요.

import asyncio

async def task(name, delay):
    # 🔽 여기에 코드 작성

import asyncio

async def task(name, delay):
    print(f"{name} 시작")
    await asyncio.sleep(delay)
    print(f"{name} 완료")

async def main():
    await asyncio.gather(
        task("작업 1", 2),
        task("작업 2", 3),
        task("작업 3", 1)
    )

asyncio.run(main())

---

✅ 문제 3: aiohttp를 사용하여 여러 웹페이지에서 데이터를 가져오기

📌 세 개의 웹사이트에 동시에 요청을 보내고 응답 길이를 출력하세요.

import aiohttp
import asyncio

async def fetch(url):
    # 🔽 여기에 코드 작성

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.text()
            print(f"{url} 응답 길이: {len(data)}")

async def main():
    urls = ["https://example.com", "https://www.python.org", "https://www.github.com"]
    await asyncio.gather(*(fetch(url) for url in urls))

asyncio.run(main())