Python의 동시 프로그래밍 기능은 크게 발전하여 개발자에게 효율적인 병렬 코드를 작성할 수 있는 강력한 도구를 제공합니다. 저는 이러한 고급 기술을 탐구하는 데 상당한 시간을 보냈으며, 제가 얻은 통찰력을 여러분과 공유하게 되어 기쁩니다.
asyncio를 사용한 비동기 프로그래밍은 I/O 바인딩 작업의 판도를 바꾸었습니다. 이를 통해 스레딩의 오버헤드 없이 여러 작업을 동시에 처리할 수 있는 비차단 코드를 작성할 수 있습니다. 다음은 asyncio를 사용하여 여러 URL에서 동시에 데이터를 가져오는 방법에 대한 간단한 예입니다.
import asyncio
import aiohttp
async def fetch_url(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
urls = ['http://example.com', 'http://example.org', 'http://example.net']
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for url, result in zip(urls, results):
print(f"Content length of {url}: {len(result)}")
asyncio.run(main())
이 코드는 여러 코루틴을 생성하여 여러 URL에서 동시에 데이터를 가져오는 방법을 보여줍니다. asyncio.gather() 함수를 사용하면 모든 코루틴이 완료되고 결과를 수집할 때까지 기다릴 수 있습니다.
asyncio는 I/O 바인딩 작업에는 탁월하지만 CPU 바인딩 작업에는 적합하지 않습니다. 이를 위해 ThreadPoolExecutor와 ProcessPoolExecutor를 모두 제공하는 동시.futures 모듈을 살펴보겠습니다. ThreadPoolExecutor는 GIL을 해제하지 않는 I/O 바인딩 작업에 이상적인 반면, ProcessPoolExecutor는 CPU 바인딩 작업에 적합합니다.
다음은 ThreadPoolExecutor를 사용하여 여러 파일을 동시에 다운로드하는 예입니다.
import concurrent.futures
import requests
def download_file(url):
response = requests.get(url)
filename = url.split('/')[-1]
with open(filename, 'wb') as f:
f.write(response.content)
return f"Downloaded {filename}"
urls = [
'https://example.com/file1.pdf',
'https://example.com/file2.pdf',
'https://example.com/file3.pdf'
]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
future_to_url = {executor.submit(download_file, url): url for url in urls}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
except Exception as exc:
print(f"{url} generated an exception: {exc}")
else:
print(data)
이 코드는 세 명의 작업자가 있는 스레드 풀을 생성하고 각 URL에 대한 다운로드 작업을 제출합니다. as_completed() 함수를 사용하면 모든 작업이 완료될 때까지 기다리지 않고 결과가 나오는 대로 처리할 수 있습니다.
CPU 바인딩 작업의 경우 ProcessPoolExecutor를 사용하여 여러 CPU 코어를 활용할 수 있습니다. 다음은 소수를 병렬로 계산하는 예입니다.
import concurrent.futures
import math
def is_prime(n):
if n < 2:
return False
for i in range(2, int(math.sqrt(n)) + 1):
if n % i == 0:
return False
return True
def find_primes(start, end):
return [n for n in range(start, end) if is_prime(n)]
ranges = [(1, 25000), (25001, 50000), (50001, 75000), (75001, 100000)]
with concurrent.futures.ProcessPoolExecutor() as executor:
results = executor.map(lambda r: find_primes(*r), ranges)
all_primes = [prime for sublist in results for prime in sublist]
print(f"Found {len(all_primes)} prime numbers")
이 코드는 소수를 찾는 작업을 4개의 범위로 분할하고 별도의 Python 프로세스를 사용하여 병렬로 처리합니다. map() 함수는 find_primes() 함수를 각 범위에 적용하고 결과를 수집합니다.
여러 프로세스로 작업할 때 프로세스 간에 데이터를 공유해야 하는 경우가 많습니다. 다중 처리 모듈은 공유 메모리 및 대기열을 포함하여 이를 위한 여러 옵션을 제공합니다. 공유 메모리 배열을 사용하는 예는 다음과 같습니다.
from multiprocessing import Process, Array
import numpy as np
def worker(shared_array, start, end):
for i in range(start, end):
shared_array[i] = i * i
if __name__ == '__main__':
size = 10000000
shared_array = Array('d', size)
# Create 4 processes
processes = []
chunk_size = size // 4
for i in range(4):
start = i * chunk_size
end = start + chunk_size if i < 3 else size
p = Process(target=worker, args=(shared_array, start, end))
processes.append(p)
p.start()
# Wait for all processes to finish
for p in processes:
p.join()
# Convert shared array to numpy array for easy manipulation
np_array = np.frombuffer(shared_array.get_obj())
print(f"Sum of squares: {np_array.sum()}")
이 코드는 공유 메모리 배열을 생성하고 네 가지 프로세스를 사용하여 숫자의 제곱을 병렬로 계산합니다. 공유 배열을 사용하면 모든 프로세스가 동일한 메모리 공간에 쓸 수 있으므로 프로세스 간 통신이 필요하지 않습니다.
이러한 기술은 강력하지만 나름대로의 과제도 있습니다. 경쟁 조건, 교착 상태 및 과도한 컨텍스트 전환은 모두 성능과 정확성에 영향을 미칠 수 있습니다. 동시 코드를 신중하게 설계하고 필요한 경우 적절한 동기화 프리미티브를 사용하는 것이 중요합니다.
예를 들어 여러 스레드나 프로세스가 공유 리소스에 액세스해야 하는 경우 잠금을 사용하여 스레드 안전성을 보장할 수 있습니다.
import asyncio
import aiohttp
async def fetch_url(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
urls = ['http://example.com', 'http://example.org', 'http://example.net']
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for url, result in zip(urls, results):
print(f"Content length of {url}: {len(result)}")
asyncio.run(main())
이 코드는 여러 스레드가 동시에 증가할 때 공유 카운터를 경쟁 조건으로부터 보호하기 위해 잠금을 사용하는 방법을 보여줍니다.
또 다른 고급 기술은 제한된 리소스에 대한 액세스를 제어하기 위해 세마포어를 사용하는 것입니다. 다음은 동시 네트워크 연결 수를 제한하는 예입니다.
import concurrent.futures
import requests
def download_file(url):
response = requests.get(url)
filename = url.split('/')[-1]
with open(filename, 'wb') as f:
f.write(response.content)
return f"Downloaded {filename}"
urls = [
'https://example.com/file1.pdf',
'https://example.com/file2.pdf',
'https://example.com/file3.pdf'
]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
future_to_url = {executor.submit(download_file, url): url for url in urls}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
except Exception as exc:
print(f"{url} generated an exception: {exc}")
else:
print(data)
이 코드는 세마포어를 사용하여 동시 네트워크 연결 수를 10개로 제한하여 네트워크나 서버에 과부하가 걸리는 것을 방지합니다.
동시 코드로 작업할 때는 예외를 올바르게 처리하는 것도 중요합니다. asyncio 모듈은 이에 유용할 수 있는 return_Exceptions 매개변수와 함께 asyncio.gather() 함수를 제공합니다.
import concurrent.futures
import math
def is_prime(n):
if n < 2:
return False
for i in range(2, int(math.sqrt(n)) + 1):
if n % i == 0:
return False
return True
def find_primes(start, end):
return [n for n in range(start, end) if is_prime(n)]
ranges = [(1, 25000), (25001, 50000), (50001, 75000), (75001, 100000)]
with concurrent.futures.ProcessPoolExecutor() as executor:
results = executor.map(lambda r: find_primes(*r), ranges)
all_primes = [prime for sublist in results for prime in sublist]
print(f"Found {len(all_primes)} prime numbers")
이 코드는 다른 작업의 실행을 중단하지 않고 동시 작업에서 예외를 처리하는 방법을 보여줍니다.
동시 프로그래밍을 더 깊이 파고들면서 이벤트 루프 및 코루틴 체이닝과 같은 고급 개념을 접하게 됩니다. 다음은 코루틴을 연결하는 방법을 보여주는 예입니다.
from multiprocessing import Process, Array
import numpy as np
def worker(shared_array, start, end):
for i in range(start, end):
shared_array[i] = i * i
if __name__ == '__main__':
size = 10000000
shared_array = Array('d', size)
# Create 4 processes
processes = []
chunk_size = size // 4
for i in range(4):
start = i * chunk_size
end = start + chunk_size if i < 3 else size
p = Process(target=worker, args=(shared_array, start, end))
processes.append(p)
p.start()
# Wait for all processes to finish
for p in processes:
p.join()
# Convert shared array to numpy array for easy manipulation
np_array = np.frombuffer(shared_array.get_obj())
print(f"Sum of squares: {np_array.sum()}")
이 코드는 세 개의 코루틴(fetch_data, process_data, save_result)을 연결하여 각 URL에 대한 파이프라인을 생성합니다. 그런 다음 asyncio.gather() 함수는 이러한 파이프라인을 동시에 실행합니다.
장기 실행 작업을 수행할 때 취소 및 시간 초과 메커니즘을 구현해야 하는 경우가 많습니다. 다음은 두 가지를 모두 보여주는 예입니다.
from threading import Lock, Thread
class Counter:
def __init__(self):
self.count = 0
self.lock = Lock()
def increment(self):
with self.lock:
self.count += 1
def worker(counter, num_increments):
for _ in range(num_increments):
counter.increment()
counter = Counter()
threads = []
for _ in range(10):
t = Thread(target=worker, args=(counter, 100000))
threads.append(t)
t.start()
for t in threads:
t.join()
print(f"Final count: {counter.count}")
이 코드는 5개의 장기 실행 작업을 시작하지만 모든 작업이 완료되는 데 시간 제한을 5초로 설정합니다. 제한 시간에 도달하면 남은 작업이 모두 취소됩니다.
결론적으로 Python의 동시 프로그래밍 기능은 효율적인 병렬 코드를 작성하기 위한 광범위한 도구와 기술을 제공합니다. asyncio를 사용한 비동기 프로그래밍부터 CPU 바인딩 작업을 위한 다중 처리에 이르기까지 이러한 고급 기술은 애플리케이션 성능을 크게 향상시킬 수 있습니다. 그러나 기본 개념을 이해하고, 각 작업에 적합한 도구를 선택하고, 공유 리소스와 잠재적인 경쟁 조건을 주의 깊게 관리하는 것이 중요합니다. 연습과 신중한 설계를 통해 Python의 동시 프로그래밍 기능을 최대한 활용하여 빠르고 확장 가능하며 반응성이 뛰어난 애플리케이션을 구축할 수 있습니다.
저희 창작물을 꼭 확인해 보세요.
인베스터 센트럴 | 스마트리빙 | 시대와 메아리 | 수수께끼의 미스터리 | 힌두트바 | 엘리트 개발자 | JS 학교
테크 코알라 인사이트 | Epochs & Echoes World | 투자자중앙매체 | 수수께끼 미스터리 매체 | 과학과 신기원 매체 | 현대 힌두트바
위 내용은 Python의 동시 프로그래밍 마스터하기: 고급 기술로 성능 향상의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
