Python 동시 프로그래밍 스레드 풀/프로세스 풀-파이썬 튜토리얼-php.cn

소개

Python 표준 라이브러리는 해당 멀티스레딩/멀티프로세스 코드를 작성하기 위한 스레딩 및 멀티프로세싱 모듈을 제공합니다. 그러나 프로젝트가 특정 규모에 도달하면 프로세스의 생성/파괴가 자주 발생합니다. 스레드는 매우 리소스 집약적입니다. 예, 지금은 시간을 위해 공간을 교환하기 위해 자체 스레드 풀/프로세스 풀을 작성해야 합니다. 그러나 Python 3.2부터 표준 라이브러리는 ThreadPoolExecutor 및 ProcessPoolExecutor라는 두 가지 클래스를 제공하는 concurrent.futures 모듈을 제공합니다. 이 모듈은 스레딩 및 다중 처리의 추가 추상화를 실현하고 스레드 풀/프로세스 작성을 용이하게 합니다. 풀은 직접적인 지원을 제공합니다.

Executor와 Future

concurrent.futures 모듈은 Executor를 기반으로 하며 추상 클래스이므로 직접 사용할 수 없습니다. 그러나 이 클래스가 제공하는 두 하위 클래스 ThreadPoolExecutor 및 ProcessPoolExecutor는 이름에서 알 수 있듯이 각각 스레드 풀 및 프로세스 풀 코드를 생성하는 데 사용됩니다. 해당 작업을 스레드 풀/프로세스 풀에 직접 넣을 수 있으며, 교착 상태를 걱정하기 위해 대기열을 유지할 필요가 없습니다. 스레드 풀/프로세스 풀이 자동으로 이를 예약합니다.

FutureJava와 nodejs 프로그래밍 경험이 있는 친구들이라면 이 개념이 익숙할 거라 믿습니다. 미래에 완성되는 작업으로 이해하시면 됩니다. 이는 비동기 프로그래밍의 기본입니다. 예를 들어 queue.get을 작동하면 결과가 반환되기를 기다리기 전에 차단이 발생하고 CPU는 다른 작업을 수행할 수 없습니다. Future는 대기 기간 동안 작업을 완료하는 데 도움이 됩니다. Python의 비동기 IO에 대해서는 이 기사를 읽은 후 내 Python 동시 프로그래밍 코루틴/비동기 IO를 참조할 수 있습니다.

p.s: 아직도 Python2.x를 고수하고 있다면 futures 모듈을 먼저 설치하세요.

pip install future

submit을 사용하여 스레드 풀/프로세스 풀 운영

먼저 다음 코드를 통해 스레드 풀의 개념을 이해해 봅시다

# example1.py
from Concurrent.futures import ThreadPoolExecutor
import time
def return_future_result(message):
time.sleep(2)
return message
pool = ThreadPoolExecutor(max_workers=2) # 최대 2개의 작업을 수용할 수 있는 풀 생성 Thread pool
future1 = pool.submit(return_future_result, ("hello")) # 스레드 풀에 작업 추가
future2 = pool.submit(return_future_result, ("world")) # 스레드 풀 A에 추가 task
print(future1.done()) # task1이 종료되는지 확인
time.sleep(3)
print(future2.done()) # task2가 종료되는지 확인
print(future1.result ()) # task1에서 반환된 결과 확인
print(future2.result()) # task2에서 반환된 결과 확인
실행된 결과를 기준으로 분석해 보겠습니다. submit 메소드를 사용하여 스레드 풀에 작업을 추가하고 submit은 Future 객체를 반환합니다. Future 객체는 미래에 완료되는 작업으로 간단히 이해될 수 있습니다. 첫 번째 print 문에서는 메인 스레드를 일시 중지하기 위해 time.sleep(3)을 사용했기 때문에 time.sleep(2) 때문에 future1이 완료되지 않았음이 분명합니다. 따라서 두 번째 print 문에 관해서는 다음과 같습니다. 스레드 풀 여기의 모든 작업이 완료되었습니다.

ziwenxie :: ~ » python example1.py
False
True
hello
world
# 위 프로그램을 실행하는 동안 ps 명령을 통해 확인할 수 있습니다. 동시에 백그라운드에서 실행 중인 세 개의 스레드
ziwenxie :: ~ » ps -eLf | grep python
ziwenxie 8361 7557 8361 3 3 19:45 pts/0 00:00:00 python example1.py
ziwenxie 8361 7557 8362 0 3 19:45 pts/0 00:00:00 python example1.py
ziwenxie 8361 7557 8363 0 3 19:45 pts/0 00:00:00 python example1.py
위 우리 It의 코드는 프로세스 풀 형태로 다시 작성할 수도 있습니다. API와 스레드 풀은 완전히 동일하므로 장황하게 설명하지 않겠습니다.

# example2.py
from Concurrent.futures import ProcessPoolExecutor
가져오기 시간
def return_future_result(message):
time.sleep(2)
반환 메시지
pool = ProcessPoolExecutor(max_workers=2)
future1 = pool.submit(return_future_result, ("hello"))
future2 = pool.submit(return_future_result, ("world"))
print(future1.done ())
time.sleep(3)
print(future2.done())
print(future1.result())
print(future2.result())
다음 실행 결과

ziwenxie :: ~ » python example2.py
False
True
hello
world
ziwenxie :: ~ » ps -eLf grep python
ziwenxie 8560 7557 8560 3 3 19:53 pts/0 00:00:00 python example2.py
ziwenxie 8560 7557 8563 0 3 19:53 pts/0 00:00:00 python example2.py
ziwenxie 8560 7557 8564 0 3 19:53 pts/0 00:00:00 python example2.py
ziwenxie 8561 8560 8561 0 1 19:53 pts/0 00:00:00 python example2.py
zi wenxie 8562 8560 8562 0 1 19:53 pts/0 00:00:00 python example2.py

맵/대기를 사용하여 스레드 풀/프로세스 풀 운영

제출 외에, Exectuor는 내장된 맵 사용법과 유사한 map 메소드도 제공합니다. 두 가지 예를 통해 차이점을 비교해 보겠습니다.

使用submit操작품回顾

# example3.py
import Concurrent.futures
import urllib.request
URLS = ['http://httpbin.org', 'http: //example.com/', 'https://api.github.com/']
def load_url(url, timeout):
with urllib.request.urlopen(url, timeout=timeout) as conn :
return conn.read()
# with 문을 사용하여 스레드를 즉시 정리할 수 있습니다
concurrent.futures.ThreadPoolExecutor(max_workers=3)를 실행자로 사용:
# 시작 작업을 로드하고 해당 URL로 각 미래를 표시합니다
future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
for future in Concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
exc 예외:
print('%r이 예외를 생성했습니다: %s' % (url, ex) )
else:
print('%r 페이지는 %d바이트입니다' % (url, len(data)))
从运行结果可以看出，as_completed저 URLS列表元素的顺序返回的。

ziwenxie :: ~ » python example3.py
'http://example.com/' 페이지는 1270바이트입니다
'https://api.github .com/' 페이지는 2039바이트
'http://httpbin.org' 페이지는 12150바이트

使용 맵

# example4.py
가져오기 동시 .futures
import urllib.request
URLS = ['http://httpbin.org', 'http://example.com/', 'https://api.github.com/']
def load_url(url):
with urllib.request.urlopen(url, timeout=60) as conn:
return conn.read()
# with 문을 사용하여 스레드가 다음과 같은지 확인할 수 있습니다. 즉시 정리
concurrent.futures.ThreadPoolExecutor(max_workers=3)를 실행자로 사용:
url의 경우 zip(URLS, executor.map(load_url, URLS))의 데이터:
print('%r 페이지는 %d바이트입니다'%(url, len(data)))
从运行结果可以看洁,map是按光URLS列表元素的顺序返回的直观，我们可以根据具体的需求任选一种。

ziwenxie :: ~ » python example4.py
'http://httpbin.org' 페이지는 12150바이트입니다
'http: //example.com/' 페이지는 1270바이트입니다
'https://api.github.com/' 페이지는 2039바이트입니다

第三种选择wait

대기 방법은 튜플(원열), 튜플 중형 집합(합), 一个是완료(已完成成), 외부 一个是미완성(未完成成)입니다.获得更大자동 속도, 它接收三个参数FIRST_COMPLETED, FIRST_EXCEPTION 및 ALL_COMPLETE, 默认设置为ALL_COMPLETED.

저희는 하단 인터페이스에서 来看一下 3个参数的区别

동시.선물 가져오기 ThreadPoolExecutor, wait, as_completed
from time import sleep
from random import randint
def return_after_random_secs(num):
sleep(randint(1, 5))
return "{}의 반환" .format(num)
pool = ThreadPoolExecutor(5)
futures = []
for x in range(5):
futures.append(pool.submit(return_after_random_secs, x))
print(wait(futures))
# print(wait(futures, timeout=None, return_when='FIRST_COMPLETED'))
如果采用默认的ALL_COMPLETED,程序会阻塞直到线程池里面的所有任务都설립 。

ziwenxie :: ~ » python example5.py
DoneAndNotDoneFutures(done={
,
<0x7f0b06373898의 미래 상태=완료 반환 str>,
<0x7f0b06352ba8의 미래 상태=완료 반환 str>,
<0x7f0b06373b00의 미래 상태=완료 반환 str>}, not_done=set())
이 FIRST_COMPLETED가 완료되었습니다. 선물(완료= {
<0x7f84109edb00의 미래 상태=완료됨 반환된 문자열>,
<0x7f840e2e9320의 미래 상태=완료됨 반환된 문자열>,
<0x7f840f25ccc0의 미래 상태=완료됨 반환된 문자열>},
not_done={<0x7f840e2e9ba8의 미래 상태=실행 중>,
<0x7f840e2e9940의 미래 상태=실행 중>})

思考题