AsyncIO 라이브러리

- 개요

병렬 처리 관점에서 I/O 작업은 I/O 작업 중에서 다른 작업을 할 수 있는지 여부에 따라 다음과 같이 구분할 수 있다

• Blocking I/O : I/O 작업이 완료될 때까지 스레드의 실행을 멈추고 대기
• Non - Blocking I/O : I/O 작업 완료 여부와 관계없이 즉시 제어권을 반환하여 다른 작업을 계속 수행

 

파이썬의 AsyncIO 라이브러리에서는 내부적으로 async / await 구문을 사용하여 동시성 코드를 작성할 수 있다

또, BeautifulSoup 라이브러리를 사용하면 XML이나 HTML 파일 안의 데이터를 손쉽게 파싱할 수 있다

따라서 2개의 라이브러리를 사용하여 Non-Blocking I/O 예제를 작성한다

 

- 전역 변수

...

# 실행 시작 시간
start = timeit.default_timer()

# 데이터를 파싱할 url
urls = ['http://daum.net', 'https://naver.com', 'https://www.tistory.com/']

...

 

코드를 수행하는 시작 시간을 start 변수에 저장한다

또, 데이터를 파싱할 웹사이트의 주소들을 urls 시퀀스에 저장한다

 

- 작업 함수

...

async def fetch(url, executor):
    # 스레드 이름 출력
    print(f"Current Thread Name : {threading.current_thread().name} / Begin : {url}")

    # 실행
    result = await loop.run_in_executor(executor, urlopen, url)

    # BeutifulSoup 라이브러리를 사용하여 HTML 문서의 <title> 태그 파싱
    soup = BeautifulSoup(result.read(), "html.parser")
    result_data = soup.title

    # 스레드 이름 출력
    print(f"Current Thread Name : {threading.current_thread().name} / End : {url}")

    # 결과 반환
    return result_data
    
...

 

비동기로 실행하며 코루틴을 반환하는 async 함수인 fetch 함수를 위와 같이 정의한다

해당 함수는 데이터를 파싱할 url과 해당 작업을 수행할 스레드를 인자로 받는다

구성은 다음과 같다

• 현재 작업을 수행하는 스레드의 이름 + 파싱 작업중인 url 주소 출력
• loop.run_in_executor 함수를 사용하여 개별 스레드에서 블로킹 작업을 비동기적으로 처리
  • urlopen : HTTP에 요청을 보내고 응답을 반환하는 블로킹 작업
• BeautifulSoup 라이브러리를 사용하여 HTML 문서를 파싱
• 파싱된 결과를 반환

 

- 작업 함수를 호출하는 outer 함수

...

async def main():
    # 스레드 풀 생성(non-block)
    executor = ThreadPoolExecutor(max_workers=10)

    # future 객체를 모아 gather에서 실행
    # 각각의 인덱스별로 스레드를 할당하여 작업
    futures = [
        asyncio.ensure_future(fetch(url, executor)) for url in urls
    ]

    result = await asyncio.gather(*futures)

    print(f"Result : {result}")
    
...

 

마찬가지로 async 함수인 main 함수를 위와 같이 선언한다

해당 함수는 인자를 받지 않으며 구성은 다음과 같다

• ThreadPoolExecutor를 사용하여 작업을 수행할 스레드를 생성
• urls 전역변수에 저장된 url을 대상으로 생성한 스레드를 통해 fetch 함수를 수행
  • 수행한 결과로 생성된 future 객체를 futures에 저장
• asyncio.gather 함수를 호출하여 여러 future를 전부 실행하고, 완료될때까지 대기
• 작업 완료 결과를 언패킹하여 result에 저장하고, 결과 출력

 

- 메인 로직

...

if __name__ == '__main__':
    # 루프 초기화
    loop = asyncio.get_event_loop()

    # 작업 완료까지 대기
    loop.run_until_complete(main())

    # 수행 시간 계산
    duration = timeit.default_timer() - start
    print(f"Total Time : {duration}")

 

최종 메인 로직은 위와 같이 구성하였다

• asyncio.get_event_loop 함수를 호출하여 비동기 함수들을 실행 + 스케줄링하는 이벤트 루프 생성
• 이벤트 루프에서 run_until_complete 함수를 호출하여 작업 완료까지 대기
• 최종 로직 수행 시간을 계산하여 출력
•  

 

- 최종 실행 결과