[asyncio python] Consumer - Producer 간단 개념 정리

배경 지식

동기 - 비동기 개념

• 비동기 프로그래밍
  • CPU 연산 대비 DB, API와 연동 과정에서 발생하는 대기 시간이 훨씬 길다.
    -> 운영 체제 등 한테 맡기고 그 시간에 CPU가 다른 처리를 하는 것을 Non-Blocking 하다고 합니다.
• 서브 루틴
  • 메인 루틴 안에 종속된 루틴
  • 메인 루틴 안에서 서브 루틴을 실행하면 서브 루틴의 코드를 실행한 뒤 다시 메인 루틴으로 돌아옴.
  • 콜 스택 ( 서브 루틴이 끝난 후 자원 모두 회수 )
• 코루틴
  • cooperative routine ( 서로 협력하는 관계 )
  • 서로 대등한 관계이며, 특정 시점에 상대방의 코드를 실행함.
  • 함수가 종료되지 않은 상태에서 메인 루틴의 코드를 실행한 뒤 다시 돌아와서 코루틴의 코드를 실행함.
  • 코루틴 함수도 종료되지 않았다면( 대기 상태 ), 코루틴 코드를 여러 번 실행할 수 있음.
  • 진입점(entry point)이 여러 개일 수 있음.

동기 - 비동기 실습 코드

• 동기 프로그래밍

import time
def find_users_sync(n):
    for i in range(1, n+1):
        print(f'{n}명 중 {i}번째 사용자 조회중...')
        time.sleep(1)
    print(f'총 {n}명 사용자 동기 조회 완료!')

def process_sync():
    start = time.time()
    find_users_sync(3)
    find_users_sync(2)
    find_users_sync(1)
    end = time.time()
    print(f'>>>동기 처리 총 소요 시간: {end - start}')

process_sync()

 

• 비동기 프로그래밍

import asyncio

async def find_users_async(n):
    for i in range(1, n+1):
        print(f'{n}명 중 {i}번째 사용자 조회중...')
        await asyncio.sleep(1) # 비동기 함수
    print(f'총 {n}명 사용자 비동기 조회 완료!')

async def process_async():
    start = time.time()
    await asyncio.gather(
        find_users_async(3),
        find_users_async(2),
        find_users_async(1)
    )
    end = time.time()
    print(f'>>>비동기 처리 총 소요 시간: {end - start}')

await process_async()

 

정리

위 동기 프로그래밍과 비동기 프로그래밍은 모두 스레드를 하나만 사용합니다.

 

동기 프로그래밍은 한 명이 1초짜리 걸리는 작업을 6번 수행하니 6초가 걸리는 것이 이해가 갑니다.

하지만 비동기에서는 스레드 하나가 6초짜리 작업을 어떻게 수행하길래 3초가 걸리는 것일까요?

 

• 코드 실행 흐름
  1. 작업 등록: 3 개의 코루틴 작업( find_users_async )를 이벤트 루프에 동시에 등록합니다. ( asyncio.gather )
  2. 작업 시작: 이벤트 루프는 세 작업을 거의 동시에 시작 시킵니다.
  3. 작업 진행
     1. find_users_async(3) 이 첫 줄을 출력하고, await asyncio.sleep(1)를 만나 제워권을 이벤트 루프로 넘깁니다.
     2. 이벤트 루프는 즉시 find_users_async(2)를 실행하고, 이 작업 역시 첫 줄 출력 후 await 을 만나 제어권을 넘깁니다.
     3. find_users_async(1)에서 또한 마찬가지의 작업이 진행 됩니다.
  4. 동시 대기: 세 작업 모두 sleep(1) 상태에서 동시에 대기 합니다. 스레드는 멈추지 않고 다른 일이 생기길 기다립니다.
  5. 작업 재개: 약 1초 후, 세 작업의 sleep이 거의 동시에 끝나므로 이벤트 루프가 각 작업을 이어서 실행합니다. 이 과정이 가장 긴 작업인 find_users_async(3)가 끝날 때 까지 반복됩니다.

 

결과적으로 총 소요 시간은 가장 오래 걸리는 작업의 시간인 약 3초가 됩니다.

이 모든 작업 전환이 하나의 스레드 안에서 일어나게 되는 것입니다.

 

 

 

 

실행 환경

• 쥬피터노트북 -> nest_asyncio 이벤트 루프 중복 실행 에러 방지

 

Case 1) Consumer 구현

import asyncio
import nest_asyncio

nest_asyncio.apply()

queue = asyncio.Queue(5)

async def consumer():
    while True:
        data = await queue.get()
        await asyncio.sleep(1)
        print(data)
        if data is None:
            break
    print("consumer end")

tasks = []
for i in range(2): # Consumer 만 대기중
    tasks.append(asyncio.create_task(consumer()))

asyncio.gather(*tasks)

• 5개의 데이터가 들어갈 수 있는 큐가 만들어졌습니다. 일종의 작업 대기열 입니다.
• Consumer는 2 개를 만들어 넣었습니다.
• data가 None이면 "consumer end"가 출력됩니다. 여기서 None이 Consumer에게 있어서 종료 신호(POISON PILL) 입니다.
• await가 앞에 안걸려 있으므로 task들은 현재 실행 대기 상태에서 끝나고 다음 셀로 넘어가게 됩니다.

 

for i in range(10):
    await queue.put(i)
    print(f'{i}번째 넣음')
for i in range(5):
    await queue.put(None)
    print(f'{i}번째 None 넣음')

• Consumer가 2 이므로 작업이 2개씩 실행되는 것처럼 보입니다.
• Queue에 9까지 데이터를 넣고, 그 이후에는 종료 시그널 None을 넣었습니다.
• 컨슈머는 2개이기 때문에 위에 셀에서 보시면 2개 None이 나오면서 작업이 종료되었고, 나머지 None은 Queue에 들어가있는 상태 입니다.

 

Case 2) Producer - Consumer 구현

import asyncio
import nest_asyncio

nest_asyncio.apply()

queue = asyncio.Queue(5)

async def consumer():
    while True:
        data = await queue.get()
        await asyncio.sleep(1)
        print(data)
        if data is None:
            break
    print("consumer end")


async def producer():
    for i in range(10):
        await queue.put(i)
    await queue.put(None)

tasks = []
tasks.append(asyncio.create_task(producer()))
for i in range(2): # Consumer 만 대기중
    tasks.append(asyncio.create_task(consumer()))

await asyncio.gather(*tasks)

 

밖에서 for 문으로 따로 데이터를 주던 녀석을 Producer로 구현하였습니다.

하나의 Producer 와 두 개의 Consumer가 구현 되었습니다.

 

보시면 Producer가 주던 10개의 테스크를 전부 마치고( 2개씩 실행되는 것처럼 보였습니다. ), 

Producer가 Queue에 None을 주며 작업을 끝냈고, 이 시그널을 받고 Consumer 하나의 작업이 끝난걸 확인할 수 있습니다.

그리고 Consumer는 두 개이기 때문에 작업이 끝나지 않는 모습 입니다.

 

Queue에 데이터가 들어올 때 까지 남은 Consumer는 영원히 작업 대기 중이겠지요.

None이 들어올 때 까지 영원히 종료되지도 않을꺼고요.

 

너무나 안타까워서 커널을 꺼줬습니다.

 

 

Producer 또한 Queue가 꽉 차게 되면, Consumer가 꺼내줄 때까지 작업을 종료하지 않고, Queue(작업 대기열)이 빌 때까지 영원히 대기하게 됩니다.

 

지금은 Consumer가 전부 작업을 처리되었기 때문에 Producer가 종료된 것을 확인할 수 있습니다.

 

이처럼 Producer - Consumer 작업을 진행할 때는 자원 관리를 잘 해주어야 합니다.