Python asyncio 동시성 제어 완벽 가이드: Semaphore, Lock, Queue로 안전한 비동기 프로그래밍 구현하기

import asyncio
import aiohttp

async def fetch_data(session, url, semaphore):
    async with semaphore:  # 세마포어 획득
        async with session.get(url) as response:
            return await response.text()

async def main():
    semaphore = asyncio.Semaphore(5)  # 최대 5개 동시 실행
    urls = [f"https://api.example.com/data/{i}" for i in range(100)]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_data(session, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks)

    print(f"총 {len(results)}개 요청 완료")

asyncio.run(main())

import asyncio

class Counter:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()

    async def increment(self):
        async with self.lock:  # Lock 획득
            # 임계 영역: 원자적 실행 보장
            current = self.value
            await asyncio.sleep(0.001)  # I/O 시뮬레이션
            self.value = current + 1

async def worker(counter, name):
    for _ in range(1000):
        await counter.increment()
    print(f"{name} 완료")

async def main():
    counter = Counter()
    await asyncio.gather(
        worker(counter, "Worker-1"),
        worker(counter, "Worker-2"),
        worker(counter, "Worker-3")
    )
    print(f"최종 값: {counter.value}")  # 3000 (정확함)

asyncio.run(main())

import asyncio
import aiohttp

async def producer(queue, urls):
    """URL을 큐에 추가"""
    for url in urls:
        await queue.put(url)
        print(f"큐에 추가: {url}")

    # 종료 신호 (워커 수만큼)
    for _ in range(3):
        await queue.put(None)

async def consumer(queue, name):
    """큐에서 URL을 가져와 처리"""
    async with aiohttp.ClientSession() as session:
        while True:
            url = await queue.get()
            if url is None:  # 종료 신호
                queue.task_done()
                break

            try:
                async with session.get(url) as response:
                    print(f"[{name}] {url} 처리 완료 (상태: {response.status})")
            finally:
                queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=10)  # 최대 10개 항목
    urls = [f"https://httpbin.org/delay/{i}" for i in range(20)]

    # Producer 1개, Consumer 3개 동시 실행
    await asyncio.gather(
        producer(queue, urls),
        consumer(queue, "Worker-1"),
        consumer(queue, "Worker-2"),
        consumer(queue, "Worker-3")
    )

    await queue.join()  # 모든 작업 완료 대기
    print("모든 작업 완료")

asyncio.run(main())

import asyncio
import aiohttp

class SafeCrawler:
    def __init__(self, max_concurrent=5):
        self.queue = asyncio.Queue()
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results_lock = asyncio.Lock()
        self.results = []

    async def fetch(self, session, url):
        async with self.semaphore:  # 동시 요청 제한
            async with session.get(url) as response:
                data = await response.text()

                async with self.results_lock:  # 결과 저장 시 Lock
                    self.results.append({"url": url, "size": len(data)})

    async def worker(self, session):
        while True:
            url = await self.queue.get()
            if url is None:
                self.queue.task_done()
                break
            try:
                await self.fetch(session, url)
            finally:
                self.queue.task_done()

    async def crawl(self, urls, num_workers=3):
        async with aiohttp.ClientSession() as session:
            # 워커 시작
            workers = [asyncio.create_task(self.worker(session)) 
                      for _ in range(num_workers)]

            # URL 큐에 추가
            for url in urls:
                await self.queue.put(url)

            # 종료 신호
            for _ in range(num_workers):
                await self.queue.put(None)

            await self.queue.join()
            await asyncio.gather(*workers)

        return self.results

async def main():
    crawler = SafeCrawler(max_concurrent=3)
    urls = [f"https://httpbin.org/delay/{i%3}" for i in range(10)]
    results = await crawler.crawl(urls, num_workers=2)
    print(f"총 {len(results)}개 페이지 크롤링 완료")

asyncio.run(main())