Python 멀티스레딩 및 멀티프로세싱 튜토리얼

게시 됨: 2022-03-11
참고: Python 3.5의 출현 이후에만 사용할 수 있는 async/await를 포함하여 몇 가지 대체 기술을 보여 달라는 인기 있는 요청에 따라 기사 끝에 몇 가지 업데이트를 추가했습니다. 즐기다!

Python을 비판하는 토론에서는 Python 코드의 여러 스레드가 동시에 실행되는 것을 방지하는 전역 인터프리터 잠금(GIL이라고도 함)으로 알려진 것을 손가락으로 가리키며 다중 스레드 작업에 Python을 사용하는 것이 얼마나 어려운지에 대해 종종 이야기합니다. 이로 인해 Python 다중 스레딩 모듈은 Python 개발자가 아니고 C++ 또는 Java와 같은 다른 언어를 사용하는 경우 예상대로 작동하지 않습니다. 특정 사항을 고려하는 한 Python에서 동시에 또는 병렬로 실행되고 결과 성능에 큰 차이를 만드는 코드를 계속 작성할 수 있음을 분명히 해야 합니다. 아직 읽지 않았다면 여기 Toptal Engineering Blog에서 Eqbal Quran의 Ruby 동시성 및 병렬 처리에 대한 기사를 살펴보시기 바랍니다.

이 Python 동시성 자습서에서는 Imgur에서 가장 인기 있는 이미지를 다운로드하는 작은 Python 스크립트를 작성합니다. 이미지를 순차적으로 다운로드하거나 한 번에 하나씩 다운로드하는 버전으로 시작합니다. 전제 조건으로 Imgur에 애플리케이션을 등록해야 합니다. 이미 Imgur 계정이 없다면 먼저 계정을 만드십시오.

이 스레딩 예제의 스크립트는 Python 3.6.4에서 테스트되었습니다. 일부 변경 사항과 함께 Python 2에서도 실행되어야 합니다. urllib는 이 두 버전의 Python 사이에서 가장 많이 변경되었습니다.

Python 멀티스레딩 시작하기

download.py 라는 Python 모듈을 만드는 것으로 시작하겠습니다. 이 파일에는 이미지 목록을 가져오고 다운로드하는 데 필요한 모든 기능이 포함됩니다. 이러한 기능을 세 개의 개별 기능으로 나눕니다.

  • get_links
  • download_link
  • setup_download_dir

세 번째 기능인 setup_download_dir 은 다운로드 대상 디렉토리가 아직 존재하지 않는 경우 이를 생성하는 데 사용됩니다.

Imgur의 API는 클라이언트 ID와 함께 Authorization 헤더를 포함하는 HTTP 요청을 요구합니다. Imgur에 등록한 애플리케이션의 대시보드에서 이 클라이언트 ID를 찾을 수 있으며 응답은 JSON으로 인코딩됩니다. Python의 표준 JSON 라이브러리를 사용하여 디코딩할 수 있습니다. 이미지 다운로드는 URL로 이미지를 가져와 파일에 쓰기만 하면 되므로 훨씬 더 간단한 작업입니다.

스크립트는 다음과 같습니다.

 import json import logging import os from pathlib import Path from urllib.request import urlopen, Request logger = logging.getLogger(__name__) types = {'image/jpeg', 'image/png'} def get_links(client_id): headers = {'Authorization': 'Client-ID {}'.format(client_id)} req = Request('https://api.imgur.com/3/gallery/random/random/', headers=headers, method='GET') with urlopen(req) as resp: data = json.loads(resp.read().decode('utf-8')) return [item['link'] for item in data['data'] if 'type' in item and item['type'] in types] def download_link(directory, link): download_path = directory / os.path.basename(link) with urlopen(link) as image, download_path.open('wb') as f: f.write(image.read()) logger.info('Downloaded %s', link) def setup_download_dir(): download_dir = Path('images') if not download_dir.exists(): download_dir.mkdir() return download_dir

다음으로 이러한 기능을 사용하여 이미지를 하나씩 다운로드하는 모듈을 작성해야 합니다. 이 single.py 의 이름을 지정합니다. 여기에는 Imgur 이미지 다운로더의 순진한 첫 번째 버전의 주요 기능이 포함됩니다. 모듈은 환경 변수 IMGUR_CLIENT_ID 에서 Imgur 클라이언트 ID를 검색합니다. setup_download_dir 을 호출하여 다운로드 대상 디렉토리를 생성합니다. 마지막으로 get_links 함수를 사용하여 이미지 목록을 가져오고 모든 GIF 및 앨범 URL을 필터링한 다음 download_link 를 사용하여 각 이미지를 다운로드하고 디스크에 저장합니다. single.py 는 다음과 같습니다.

 import logging import os from time import time from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def main(): ts = time() client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!") download_dir = setup_download_dir() links = get_links(client_id) for link in links: download_link(download_dir, link) logging.info('Took %s seconds', time() - ts) if __name__ == '__main__': main()

내 노트북에서 이 스크립트는 91개의 이미지를 다운로드하는 데 19.4초가 걸렸습니다. 이 숫자는 사용 중인 네트워크에 따라 다를 수 있습니다. 19.4초는 그다지 길지 않지만 더 많은 사진을 다운로드하려면 어떻게 해야 할까요? 90개가 아닌 900개 이미지일 수 있습니다. 사진당 평균 0.2초로 900개 이미지는 약 3분이 걸립니다. 9000장의 사진은 30분이 소요됩니다. 좋은 소식은 동시성 또는 병렬성을 도입하여 이를 극적으로 가속화할 수 있다는 것입니다.

모든 후속 코드 예제는 해당 예제와 관련된 새롭고 특정한 가져오기 문만 표시합니다. 편의를 위해 이 모든 Python 스크립트는 이 GitHub 리포지토리에서 찾을 수 있습니다.

Python의 동시성 및 병렬성: 스레딩 예제

스레딩은 Python 동시성과 병렬성을 달성하기 위한 가장 잘 알려진 접근 방식 중 하나입니다. 스레딩은 일반적으로 운영 체제에서 제공하는 기능입니다. 스레드는 프로세스보다 가볍고 동일한 메모리 공간을 공유합니다.

Python 멀티스레딩 메모리 모델

이 Python 스레딩 예제에서는 single.py 를 대체할 새 모듈을 작성합니다. 이 모듈은 8개의 스레드로 구성된 풀을 생성하여 메인 스레드를 포함하여 총 9개의 스레드를 만듭니다. 내 컴퓨터에는 8개의 CPU 코어가 있고 코어당 하나의 작업자 스레드는 한 번에 실행할 스레드 수에 대해 좋은 숫자로 보였기 때문에 8개의 작업자 스레드를 선택했습니다. 실제로 이 숫자는 동일한 시스템에서 실행되는 다른 응용 프로그램 및 서비스와 같은 다른 요소를 기반으로 훨씬 더 신중하게 선택됩니다.

Python Thread 클래스의 후손인 DownloadWorker 라는 새 클래스가 있다는 점을 제외하고는 이전 클래스와 거의 동일합니다. 무한 루프를 실행하는 run 메서드가 재정의되었습니다. 모든 반복에서 self.queue.get() 을 호출하여 스레드로부터 안전한 큐에서 URL을 가져오려고 시도합니다. 작업자가 처리할 항목이 대기열에 있을 때까지 차단됩니다. 작업자가 대기열에서 항목을 받으면 이전 스크립트에서 이미지를 images 디렉터리로 다운로드하는 데 사용한 것과 동일한 download_link 메서드를 호출합니다. 다운로드가 완료된 후 작업자는 해당 작업이 완료되었음을 큐에 알립니다. 대기열이 대기열에 추가된 작업 수를 추적하기 때문에 이는 매우 중요합니다. 워커가 작업을 완료했다는 신호를 보내지 않으면 queue.join() 호출은 메인 스레드를 영원히 차단합니다.

 import logging import os from queue import Queue from threading import Thread from time import time from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class DownloadWorker(Thread): def __init__(self, queue): Thread.__init__(self) self.queue = queue def run(self): while True: # Get the work from the queue and expand the tuple directory, link = self.queue.get() try: download_link(directory, link) finally: self.queue.task_done() def main(): ts = time() client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!") download_dir = setup_download_dir() links = get_links(client_id) # Create a queue to communicate with the worker threads queue = Queue() # Create 8 worker threads for x in range(8): worker = DownloadWorker(queue) # Setting daemon to True will let the main thread exit even though the workers are blocking worker.daemon = True worker.start() # Put the tasks into the queue as a tuple for link in links: logger.info('Queueing {}'.format(link)) queue.put((download_dir, link)) # Causes the main thread to wait for the queue to finish processing all the tasks queue.join() logging.info('Took %s', time() - ts) if __name__ == '__main__': main()

이전에 사용한 것과 동일한 시스템에서 이 Python 스레딩 예제 스크립트를 실행하면 다운로드 시간이 4.1초가 됩니다! 이는 이전 예보다 4.7배 빠릅니다. 이것이 훨씬 빠르지만 GIL로 인해 이 프로세스 전체에서 한 번에 하나의 스레드만 실행되고 있었다는 점을 언급할 가치가 있습니다. 따라서 이 코드는 동시적이지만 병렬적이지 않습니다. 여전히 더 빠른 이유는 이것이 IO 바운드 작업이기 때문입니다. 프로세서는 이러한 이미지를 다운로드하는 동안 거의 땀을 흘리지 않으며 대부분의 시간은 네트워크를 기다리는 데 소비됩니다. 이것이 Python 멀티스레딩이 속도를 크게 높일 수 있는 이유입니다. 프로세서는 스레드 중 하나가 작업을 수행할 준비가 될 때마다 스레드 간에 전환할 수 있습니다. Python 또는 GIL과 함께 다른 해석된 언어에서 스레딩 모듈을 사용하면 실제로 성능이 저하될 수 있습니다. 코드가 gzip 파일 압축 해제와 같은 CPU 바운드 작업을 수행하는 경우 threading 모듈을 사용하면 실행 시간이 느려집니다. CPU 바운드 작업과 진정한 병렬 실행을 위해 멀티프로세싱 모듈을 사용할 수 있습니다.

사실상 참조 Python 구현인 CPython에는 GIL이 있지만 모든 Python 구현에 해당하는 것은 아닙니다. 예를 들어 .NET 프레임워크를 사용하는 Python 구현인 IronPython에는 GIL이 없으며 Java 기반 구현인 Jython도 없습니다. 여기에서 작동하는 Python 구현 목록을 찾을 수 있습니다.

관련: Toptal 개발자의 Python 모범 사례 및 팁

Python 예제 2의 동시성 및 병렬성: 여러 프로세스 생성

multiprocessing 모듈은 Python 스레딩 예제와 같은 클래스를 추가할 필요가 없기 때문에 스레딩 모듈보다 넣기가 더 쉽습니다. 우리가 해야 할 유일한 변경은 main 함수에 있습니다.

Python 다중 처리 튜토리얼: 모듈

여러 프로세스를 사용하기 위해 멀티프로세싱 Pool 을 생성합니다. 제공하는 map 메서드를 사용하여 URL 목록을 풀에 전달하면 풀에서 8개의 새 프로세스가 생성되고 각 프로세스를 사용하여 이미지를 병렬로 다운로드합니다. 이것은 진정한 병렬 처리이지만 비용이 따릅니다. 스크립트의 전체 메모리는 생성되는 각 하위 프로세스에 복사됩니다. 이 간단한 예에서는 큰 문제가 아니지만 사소하지 않은 프로그램의 경우 심각한 오버헤드가 되기 쉽습니다.

 import logging import os from functools import partial from multiprocessing.pool import Pool from time import time from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logging.getLogger('requests').setLevel(logging.CRITICAL) logger = logging.getLogger(__name__) def main(): ts = time() client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!") download_dir = setup_download_dir() links = get_links(client_id) download = partial(download_link, download_dir) with Pool(4) as p: p.map(download, links) logging.info('Took %s seconds', time() - ts) if __name__ == '__main__': main()

Python 예제 3의 동시성 및 병렬 처리: 여러 작업자에 배포

스레딩 및 멀티프로세싱 모듈은 개인용 컴퓨터에서 실행되는 스크립트에 적합하지만, 다른 컴퓨터에서 작업을 수행하거나 한 컴퓨터의 CPU가 할 수 있는 것 이상으로 확장해야 하는 경우에는 어떻게 해야 합니까? 핸들? 이에 대한 훌륭한 사용 사례는 웹 애플리케이션을 위한 장기 실행 백엔드 작업입니다. 장기 실행 작업이 있는 경우 애플리케이션 코드의 나머지 부분을 실행해야 하는 동일한 시스템에서 많은 하위 프로세스 또는 스레드를 실행하고 싶지 않을 것입니다. 이렇게 하면 모든 사용자의 애플리케이션 성능이 저하됩니다. 좋은 것은 이러한 작업을 다른 컴퓨터나 다른 많은 컴퓨터에서 실행할 수 있다는 것입니다.

이 작업을 위한 훌륭한 Python 라이브러리는 매우 간단하지만 강력한 라이브러리인 RQ입니다. 먼저 라이브러리를 사용하여 함수와 해당 인수를 대기열에 넣습니다. 이것은 함수 호출 표현을 피클링한 다음 Redis 목록에 추가합니다. 작업을 대기열에 넣는 것이 첫 번째 단계이지만 아직 아무 작업도 수행하지 않습니다. 또한 해당 작업 대기열에서 수신 대기할 작업자가 한 명 이상 필요합니다.

RQ Python 큐 라이브러리 모델

첫 번째 단계는 컴퓨터에 Redis 서버를 설치 및 실행하거나 실행 중인 Redis 서버에 액세스하는 것입니다. 그 후에는 기존 코드에 약간의 변경만 있습니다. 먼저 RQ 큐의 인스턴스를 만들고 redis-py 라이브러리에서 Redis 서버의 인스턴스를 전달합니다. 그런 다음 download_link 메소드를 호출하는 대신 q.enqueue(download_link, download_dir, link) 를 호출합니다. enqueue 메소드는 함수를 첫 번째 인수로 취한 다음 작업이 실제로 실행될 때 다른 인수 또는 키워드 인수가 해당 함수에 전달됩니다.

우리가 해야 할 마지막 단계는 일부 작업자를 시작하는 것입니다. RQ는 기본 대기열에서 작업자를 실행하는 편리한 스크립트를 제공합니다. 터미널 창에서 rqworker 를 실행하면 기본 대기열에서 수신 대기 중인 작업자가 시작됩니다. 현재 작업 디렉토리가 스크립트가 있는 위치와 동일한지 확인하십시오. 다른 대기열을 수신하려면 rqworker queue_name 을 실행하면 됩니다. 그러면 대기열이 명명된 대기열을 수신합니다. RQ의 가장 큰 장점은 Redis에 연결할 수 있는 한 원하는 만큼 다양한 시스템에서 원하는 만큼 작업자를 실행할 수 있다는 것입니다. 따라서 애플리케이션이 성장함에 따라 확장하기가 매우 쉽습니다. 다음은 RQ 버전의 소스입니다.

 import logging import os from redis import Redis from rq import Queue from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logging.getLogger('requests').setLevel(logging.CRITICAL) logger = logging.getLogger(__name__) def main(): client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!") download_dir = setup_download_dir() links = get_links(client_id) q = Queue(connection=Redis(host='localhost', port=6379)) for link in links: q.enqueue(download_link, download_dir, link) if __name__ == '__main__': main()

그러나 RQ가 유일한 Python 작업 대기열 솔루션은 아닙니다. RQ는 사용하기 쉽고 간단한 사용 사례를 매우 잘 다루지만 더 고급 옵션이 필요한 경우 다른 Python 3 대기열 솔루션(예: Celery)을 사용할 수 있습니다.

Python 멀티스레딩 대 멀티프로세싱

코드가 IO 바운드인 경우 Python의 다중 처리 및 다중 스레딩이 모두 작동합니다. 멀티프로세싱은 스레딩보다 그냥 넣기가 더 쉽지만 메모리 오버헤드가 더 높습니다. 코드가 CPU 바운드인 경우 다중 처리가 더 나은 선택일 가능성이 높습니다. 특히 대상 시스템에 다중 코어 또는 CPU가 있는 경우에는 더욱 그렇습니다. 웹 애플리케이션의 경우 작업을 여러 시스템으로 확장해야 하는 경우 RQ가 더 좋습니다.

관련 항목: 고급화: Python 프로그래머가 저지르는 가장 일반적인 실수 10가지 방지

업데이트

파이썬 concurrent.futures

원본 기사에서 다루지 않은 Python 3.2 이후의 새로운 기능은 concurrent.futures 패키지입니다. 이 패키지는 Python에서 동시성과 병렬성을 사용하는 또 다른 방법을 제공합니다.

원래 기사에서 나는 파이썬의 다중 처리 모듈이 스레딩 모듈보다 기존 코드에 더 쉽게 들어갈 것이라고 언급했습니다. 이는 Python 3 스레딩 모듈이 Thread 클래스를 서브클래싱하고 스레드가 작업을 모니터링할 Queue 을 생성해야 했기 때문입니다.

concurrent.futures.ThreadPoolExecutor를 사용하면 Python 스레딩 예제 코드가 다중 처리 모듈과 거의 동일합니다.

 import logging import os from concurrent.futures import ThreadPoolExecutor from functools import partial from time import time from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def main(): client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!") download_dir = setup_download_dir() links = get_links(client_id) # By placing the executor inside a with block, the executors shutdown method # will be called cleaning up threads. # # By default, the executor sets number of workers to 5 times the number of # CPUs. with ThreadPoolExecutor() as executor: # Create a new partially applied function that stores the directory # argument. # # This allows the download_link function that normally takes two # arguments to work with the map function that expects a function of a # single argument. fn = partial(download_link, download_dir) # Executes fn concurrently using threads on the links iterable. The # timeout is for the entire process, not a single call, so downloading # all images must complete within 30 seconds. executor.map(fn, links, timeout=30) if __name__ == '__main__': main()

이제 Python ThreadPoolExecutor 로 이 모든 이미지를 다운로드했으므로 이를 사용하여 CPU 바운드 작업을 테스트할 수 있습니다. 단일 스레드, 단일 프로세스 스크립트 모두에서 모든 이미지의 축소판 버전을 만든 다음 다중 처리 기반 솔루션을 테스트할 수 있습니다.

Pillow 라이브러리를 사용하여 이미지 크기 조정을 처리할 것입니다.

다음은 초기 스크립트입니다.

 import logging from pathlib import Path from time import time from PIL import Image logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def create_thumbnail(size, path): """ Creates a thumbnail of an image with the same name as image but with _thumbnail appended before the extension. Eg: >>> create_thumbnail((128, 128), 'image.jpg') A new thumbnail image is created with the name image_thumbnail.jpg :param size: A tuple of the width and height of the image :param path: The path to the image file :return: None """ image = Image.open(path) image.thumbnail(size) path = Path(path) name = path.stem + '_thumbnail' + path.suffix thumbnail_path = path.with_name(name) image.save(thumbnail_path) def main(): ts = time() for image_path in Path('images').iterdir(): create_thumbnail((128, 128), image_path) logging.info('Took %s', time() - ts) if __name__ == '__main__': main()

이 스크립트는 images 폴더의 경로를 반복하고 각 경로에 대해 create_thumbnail 함수를 실행합니다. 이 기능은 Pillow를 사용하여 이미지를 열고 축소판을 만들고 원본과 이름이 같지만 이름에 _thumbnail 이 추가된 더 작은 새 이미지를 저장합니다.

총 3,600만 개의 이미지 160개에서 이 스크립트를 실행하는 데 2.32초가 걸립니다. ProcessPoolExecutor를 사용하여 속도를 높일 수 있는지 봅시다.

 import logging from pathlib import Path from time import time from functools import partial from concurrent.futures import ProcessPoolExecutor from PIL import Image logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def create_thumbnail(size, path): """ Creates a thumbnail of an image with the same name as image but with _thumbnail appended before the extension. Eg: >>> create_thumbnail((128, 128), 'image.jpg') A new thumbnail image is created with the name image_thumbnail.jpg :param size: A tuple of the width and height of the image :param path: The path to the image file :return: None """ path = Path(path) name = path.stem + '_thumbnail' + path.suffix thumbnail_path = path.with_name(name) image = Image.open(path) image.thumbnail(size) image.save(thumbnail_path) def main(): ts = time() # Partially apply the create_thumbnail method, setting the size to 128x128 # and returning a function of a single argument. thumbnail_128 = partial(create_thumbnail, (128, 128)) # Create the executor in a with block so shutdown is called when the block # is exited. with ProcessPoolExecutor() as executor: executor.map(thumbnail_128, Path('images').iterdir()) logging.info('Took %s', time() - ts) if __name__ == '__main__': main()

create_thumbnail 메소드는 마지막 스크립트와 동일합니다. 주요 차이점은 ProcessPoolExecutor 생성입니다. executor의 map 메소드는 썸네일을 병렬로 생성하는 데 사용됩니다. 기본적으로 ProcessPoolExecutor 는 CPU당 하나의 하위 프로세스를 만듭니다. 동일한 160개의 이미지에서 이 스크립트를 실행하는 데 1.05초가 걸렸습니다. 2.2배 더 빨라졌습니다!

비동기/대기(Python 3.5 이상만 해당)

원본 기사의 댓글에서 가장 많이 요청된 항목 중 하나는 Python 3의 asyncio 모듈을 사용한 예제였습니다. 다른 예와 비교할 때, 대부분의 사람들에게 생소할 수 있는 몇 가지 새로운 Python 구문과 몇 가지 새로운 개념이 있습니다. 불행한 추가 복잡성 계층은 Python의 내장 urllib 모듈이 비동기적이지 않기 때문에 발생합니다. asyncio의 모든 이점을 얻으려면 비동기 HTTP 라이브러리를 사용해야 합니다. 이를 위해 우리는 iohttp를 사용할 것입니다.

바로 코드로 넘어가서 더 자세한 설명이 이어집니다.

 import asyncio import logging import os from time import time import aiohttp from download import setup_download_dir, get_links logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) async def async_download_link(session, directory, link): """ Async version of the download_link method we've been using in the other examples. :param session: aiohttp ClientSession :param directory: directory to save downloads :param link: the url of the link to download :return: """ download_path = directory / os.path.basename(link) async with session.get(link) as response: with download_path.open('wb') as f: while True: # await pauses execution until the 1024 (or less) bytes are read from the stream chunk = await response.content.read(1024) if not chunk: # We are done reading the file, break out of the while loop break f.write(chunk) logger.info('Downloaded %s', link) # Main is now a coroutine async def main(): client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!") download_dir = setup_download_dir() # We use a session to take advantage of tcp keep-alive # Set a 3 second read and connect timeout. Default is 5 minutes async with aiohttp.ClientSession(conn_timeout=3, read_timeout=3) as session: tasks = [(async_download_link(session, download_dir, l)) for l in get_links(client_id)] # gather aggregates all the tasks and schedules them in the event loop await asyncio.gather(*tasks, return_exceptions=True) if __name__ == '__main__': ts = time() # Create the asyncio event loop loop = asyncio.get_event_loop() try: loop.run_until_complete(main()) finally: # Shutdown the loop even if there is an exception loop.close() logger.info('Took %s seconds to complete', time() - ts)

여기에서 풀어야 할 것이 꽤 있습니다. 프로그램의 주요 진입점부터 시작하겠습니다. asyncio 모듈을 사용하여 새로운 첫 번째 작업은 이벤트 루프를 얻는 것입니다. 이벤트 루프는 모든 비동기 코드를 처리합니다. 그런 다음 루프가 완료될 때까지 실행되고 main 함수가 전달됩니다. main: async def 의 정의에 새로운 구문이 있습니다. 또한 await 및 with async 를 알 수 있습니다.

async/await 구문은 PEP492에 도입되었습니다. async def 구문은 함수를 코루틴으로 표시합니다. 내부적으로 코루틴은 Python 생성기를 기반으로 하지만 정확히 같은 것은 아닙니다. 코루틴은 제너레이터가 제너레이터 객체를 반환하는 방식과 유사한 코루틴 객체를 반환합니다. 코루틴이 있으면 await 표현식으로 결과를 얻습니다. 코루틴이 await 를 호출하면 코루틴의 실행은 awaitable이 완료될 때까지 일시 중단됩니다. 이 일시 중단을 통해 코루틴이 어떤 결과를 "기다리는" 일시 중단되는 동안 다른 작업을 완료할 수 있습니다. 일반적으로 이 결과는 데이터베이스 요청 또는 이 경우 HTTP 요청과 같은 일종의 I/O입니다.

download_link 기능은 상당히 변경되어야 했습니다. 이전에는 urllib 에 의존하여 이미지를 읽는 작업을 직접 수행했습니다. 이제 메서드가 비동기 프로그래밍 패러다임에서 제대로 작동할 수 있도록 한 번에 이미지 청크를 읽고 I/O가 완료될 때까지 기다리는 동안 실행을 일시 중단하는 while 루프를 도입했습니다. 이렇게 하면 이벤트 루프가 다운로드하는 동안 사용할 수 있는 새 데이터가 있으므로 서로 다른 이미지를 다운로드하는 과정을 반복할 수 있습니다.

하나가 있어야 함 - 가급적이면 하나만 - 명백한 방법

zen of Python은 무언가를 하는 한 가지 분명한 방법이 있어야 한다고 말하지만 Python에는 프로그램에 동시성을 도입하는 많은 방법이 있습니다. 선택하는 가장 좋은 방법은 특정 사용 사례에 따라 다릅니다. 비동기 패러다임은 스레딩 또는 멀티프로세싱에 비해 동시성이 높은 워크로드(예: 웹 서버)로 확장되지만 완전한 이점을 얻으려면 코드(및 종속성)가 비동기화되어야 합니다.

이 기사와 업데이트의 Python 스레딩 예제가 올바른 방향을 제시하여 프로그램에 동시성을 도입해야 하는 경우 Python 표준 라이브러리에서 볼 위치를 알 수 있기를 바랍니다.