Python ThreadPoolExecutor의 스레드 풀 문제를 해결하는 방법

Concept

Python에는 이미 threading 모듈이 있는데 왜 스레드 풀이 필요한가요? Python中已经有了threading模块，为什么还需要线程池呢，线程池又是什么东西呢？

以爬虫为例，需要控制同时爬取的线程数，例子中创建了20个线程，而同时只允许3个线程在运行，但是20个线程都需要创建和销毁，线程的创建是需要消耗系统资源的，有没有更好的方案呢？

其实只需要三个线程就行了，每个线程各分配一个任务，剩下的任务排队等待，当某个线程完成了任务的时候，排队任务就可以安排给这个线程继续执行。

这就是线程池的思想（当然没这么简单），但是自己编写线程池很难写的比较完美，还需要考虑复杂情况下的线程同步，很容易发生死锁。

从Python3.2开始，标准库为我们提供了concurrent.futures模块，它提供了ThreadPoolExecutor和ProcessPoolExecutor两个类，实现了对threading和multiprocessing的进一步抽象（这里主要关注线程池），不仅可以帮我们自动调度线程，还可以做到：

• 主线程可以获取某一个线程（或者任务的）的状态，以及返回值。

• 当一个线程完成的时候，主线程能够立即知道。

• 让多线程和多进程的编码接口一致。

实例

简单使用

from concurrent.futures import ThreadPoolExecutor
import time
 
# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
 
executor = ThreadPoolExecutor(max_workers=2)
# 通过submit函数提交执行的函数到线程池中，submit函数立即返回，不阻塞
task1 = executor.submit(get_html, (3))
task2 = executor.submit(get_html, (2))
# done方法用于判定某个任务是否完成
print(task1.done())
# cancel方法用于取消某个任务,该任务没有放入线程池中才能取消成功
print(task2.cancel())
time.sleep(4)
print(task1.done())
# result方法可以获取task的执行结果
print(task1.result())
 
# 执行结果
# False  # 表明task1未执行完成
# False  # 表明task2取消失败，因为已经放入了线程池中
# get page 2s finished
# get page 3s finished
# True  # 由于在get page 3s finished之后才打印，所以此时task1必然完成了
# 3     # 得到task1的任务返回值

ThreadPoolExecutor构造实例的时候，传入max_workers参数来设置线程池中最多能同时运行的线程数目。

使用submit函数来提交线程需要执行的任务（函数名和参数）到线程池中，并返回该任务的句柄（类似于文件、画图），注意submit()不是阻塞的，而是立即返回。

通过submit函数返回的任务句柄，能够使用done()方法判断该任务是否结束。上面的例子可以看出，由于任务有2s的延时，在task1提交后立刻判断，task1还未完成，而在延时4s之后判断，task1就完成了。

使用cancel()方法可以取消提交的任务，如果任务已经在线程池中运行了，就取消不了。这个例子中，线程池的大小设置为2，任务已经在运行了，所以取消失败。如果改变线程池的大小为1，那么先提交的是task1，task2还在排队等候，这是时候就可以成功取消。

使用result()方法可以获取任务的返回值。查看内部代码，发现这个方法是阻塞的。

as_completed

上面虽然提供了判断任务是否结束的方法，但是不能在主线程中一直判断啊。

有时候我们是得知某个任务结束了，就去获取结果，而不是一直判断每个任务有没有结束。

这是就可以使用as_completed方法一次取出所有任务的结果。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
 
# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
 
executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]
 
for future in as_completed(all_task):
    data = future.result()
    print("in main: get page {}s success".format(data))
 
# 执行结果
# get page 2s finished
# in main: get page 2s success
# get page 3s finished
# in main: get page 3s success
# get page 4s finished
# in main: get page 4s success

as_completed()方法是一个生成器，在没有任务完成的时候，会阻塞，在有某个任务完成的时候，会yield这个任务，就能执行for循环下面的语句，然后继续阻塞住，循环到所有的任务结束。

从结果也可以看出，先完成的任务会先通知主线程。

map

除了上面的as_completed方法，还可以使用executor.map方法，但是有一点不同。

from concurrent.futures import ThreadPoolExecutor
import time
 
# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
 
executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
 
for data in executor.map(get_html, urls):
    print("in main: get page {}s success".format(data))
# 执行结果
# get page 2s finished
# get page 3s finished
# in main: get page 3s success
# in main: get page 2s success
# get page 4s finished
# in main: get page 4s success

使用map方法，无需提前使用submit方法，map方法与python标准库中的map含义相同，都是将序列中的每个元素都执行同一个函数。

上面的代码就是对urls的每个元素都执行get_html函数，并分配各线程池。可以看到执行结果与上面的as_completed方法的结果不同，输出顺序和urls列表的顺序相同，就算2s的任务先执行完成，也会先打印出3s的任务先完成，再打印2s的任务完成。

wait

wait方法可以让主线程阻塞，直到满足设定的要求。

from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import time
 
# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
 
executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]
wait(all_task, return_when=ALL_COMPLETED)
print("main")
# 执行结果 
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main

wait方法接收3个参数，等待的任务序列、超时时间以及等待条件。

等待条件return_when默认为ALL_COMPLETED，表明要等待所有的任务都结束。

可以看到运行结果中，确实是所有任务都完成了，主线程才打印出main。

等待条件还可以设置为FIRST_COMPLETED，表示第一个任务完成就停止等待。

源码分析

cocurrent.future模块中的future크롤러를 예로 들어보겠습니다. 동시에 크롤링되는 스레드 수를 제어해야 합니다. 이 예에서는 20개의 스레드만 동시에 실행할 수 있지만 20개의 스레드만 필요합니다. 스레드 생성에는 시스템 리소스 측면에서 더 나은 솔루션이 있습니까?

실제로는 3개의 스레드만 필요합니다. 각 스레드에는 작업이 할당되고 나머지 작업은 대기하기 위해 대기열에 추가됩니다. 스레드가 작업을 완료하면 이 스레드가 계속 실행되도록 대기열에 있는 작업을 정렬할 수 있습니다. 🎜🎜이것이 스레드 풀의 아이디어이지만(물론 그렇게 간단하지는 않습니다) 스레드 풀을 혼자서 완벽하게 작성하기는 어렵습니다. 또한 복잡한 상황에서는 스레드 동기화를 고려해야 하며 교착 상태가 쉽게 발생할 수 있습니다. 발생하다. 🎜🎜Python3.2부터 표준 라이브러리는 ThreadPoolExecutor 및 ProcessPoolExecutorconcurrent.futures 모듈을 제공합니다. /code>두 클래스는 스레딩 및 다중 처리의 추가 추상화를 실현합니다(여기서는 주로 스레드 풀에 중점을 둡니다). 이 클래스는 스레드를 자동으로 예약하는 데 도움이 될 뿐만 아니라 ：🎜

• 🎜메인 스레드는 특정 스레드(또는 작업)의 상태와 반환 값을 얻을 수 있습니다. 🎜
• 🎜스레드가 완료되면 메인 스레드에서 즉시 알 수 있습니다. 🎜
• 🎜멀티 스레드 및 멀티 프로세스의 코딩 인터페이스를 일관되게 만드세요. 🎜
🎜🎜Instance🎜

사용이 간편

rrreee🎜ThreadPoolExecutor 인스턴스를 구성할 때 max_workers 매개변수를 전달하여 스레드 풀에서 동시에 실행할 수 있는 최대 스레드 수를 설정합니다. 🎜🎜제출 기능을 사용하여 스레드가 수행해야 하는 작업(함수 이름 및 매개변수)을 스레드 풀에 제출하고 작업 핸들(파일 및 도면과 유사)을 반환합니다. submit()은 차단되지 않습니다. , 그러나 즉시 반환됩니다. 🎜🎜제출 함수에서 반환된 작업 핸들을 사용하면 done() 메서드를 사용하여 작업이 종료되었는지 확인할 수 있습니다. 위의 예에서 알 수 있듯이 태스크는 2초의 지연이 있기 때문에 태스크1이 제출된 직후에는 태스크1이 완료되지 않은 것으로 판단되지만, 4초의 지연 후에는 태스크1이 완료된 것으로 판단된다. 🎜🎜제출된 작업을 취소하려면 cancel() 메서드를 사용하세요. 작업이 이미 스레드 풀에서 실행 중인 경우 취소할 수 없습니다. 이 예에서는 스레드 풀 크기가 2로 설정되어 있고 작업이 이미 실행 중이므로 취소가 실패합니다. 스레드 풀의 크기를 1로 변경하면 task1이 먼저 제출되고 task2는 여전히 대기열에서 대기 중이므로 성공적으로 취소될 수 있습니다. 🎜🎜작업의 반환 값을 얻으려면 result() 메서드를 사용하세요. 내부 코드를 살펴보면 이 메서드가 차단되고 있음을 발견했습니다. 🎜🎜as_completed🎜🎜위의 방법은 작업 완료 여부를 확인하는 방법을 제공하지만 항상 메인 스레드에서 확인할 수는 없습니다. 🎜🎜때때로 우리는 특정 작업이 끝났다는 것을 알 때 각 작업이 끝났는지 항상 판단하는 대신 결과를 얻습니다. 🎜🎜모든 작업을 한번에 조회하기 위해 as_completed 메소드를 사용한 결과입니다. 🎜rrreee🎜as_completed() 메서드는 작업이 완료되지 않으면 차단됩니다. 특정 작업이 완료되면 작업이 생성됩니다. for 루프 아래의 명령문을 실행한 다음 모든 작업이 완료될 때까지 계속 차단할 수 있습니다. 🎜🎜결과에서도 🎜먼저 완료된 작업이 메인 스레드에 먼저 통보됩니다🎜는 것을 알 수 있습니다. 🎜

map

🎜위의 as_completed 메서드 외에 executor.map 메서드를 사용할 수도 있지만 약간의 차이가 있습니다. 🎜rrreee🎜미리 submit 메서드를 사용하지 않고 map 메서드를 사용하세요. map 메서드는 python과 동일합니다. code> 표준 라이브러리. code>map은 동일한 의미를 가지며 시퀀스의 각 요소에 대해 동일한 기능을 실행합니다. 🎜🎜위 코드는 urls의 각 요소에 대해 get_html 함수를 실행하고 각 스레드 풀을 할당하는 코드입니다. 위의 as_completed 메소드의 결과와 실행 결과가 다른 것을 확인할 수 있습니다. 🎜urls🎜목록의 순서와 동일합니다. 2s 작업이 먼저 완료되고, 3s 작업이 먼저 인쇄된 다음 2s 작업이 인쇄됩니다. 🎜

wait

🎜wait 메서드는 설정된 요구 사항이 충족될 때까지 메인 스레드를 차단할 수 있습니다. 🎜rrreee🎜 wait 메서드는 대기 작업 순서, 시간 초과 시간, 대기 조건 등 3개의 매개변수를 받습니다. 🎜🎜대기 조건 return_when은 기본적으로 ALL_COMPLETED로 설정되어 모든 작업이 끝날 때까지 기다리고 있음을 나타냅니다. 🎜🎜실행 결과에서 모든 작업이 실제로 완료되었음을 확인할 수 있으며, 메인 스레드는 main을 인쇄합니다. 🎜🎜대기 조건을 FIRST_COMPLETED로 설정할 수도 있습니다. 즉, 첫 번째 작업이 완료되면 대기가 중지됩니다. 🎜🎜cocurrent.future 모듈의 🎜🎜소스 코드 분석🎜🎜future은 future 객체를 의미하며, 이는 🎜미래에 완료된 작업🎜으로 이해될 수 있으며, 이는 비동기 기본입니다. 프로그래밍의. 🎜

스레드 풀 submit() 이후 future 객체가 반환될 때 작업이 완료되지 않지만 나중에 완료됩니다. submit()之后，返回的就是这个future对象，返回的时候任务并没有完成，但会在将来完成。

也可以称之为task的返回容器，这个里面会存储task的结果和状态。

那ThreadPoolExecutor内部是如何操作这个对象的呢？

下面简单介绍ThreadPoolExecutor的部分代码：

1.init方法

init方法中主要重要的就是任务队列和线程集合，在其他方法中需要使用到。



2.submit方法

submit中有两个重要的对象，_base.Future()和_WorkItem()对象，_WorkItem()对象负责运行任务和对future对象进行设置，最后会将future对象返回，可以看到整个过程是立即返回的，没有阻塞。



3.adjust_thread_count方法

这个方法的含义很好理解，主要是创建指定的线程数。但是实现上有点难以理解，比如线程执行函数中的weakref.ref，涉及到了弱引用等概念，留待以后理解。



4._WorkItem对象

_WorkItem对象的职责就是执行任务和设置结果。这里面主要复杂的还是self.future.set_result(result)은 task

의 결과와 상태를 저장하는 작업의 반환 컨테이너라고도 할 수 있습니다.

그렇다면 ThreadPoolExecutor는 이 개체를 내부적으로 어떻게 작동합니까?

다음은 ThreadPoolExecutor 코드 일부에 대한 간략한 소개입니다.

1.init 메소드 init 메소드에서 가장 중요한 것은 작업 대기열입니다. 및 기존의 다른 방법에 필요한 스레드 수집입니다.



2 .submit 메소드 🎜🎜submit, _base.Future() 및 _WorkItem() 객체에는 두 가지 중요한 객체인 _WorkItem이 있습니다. () 객체는 작업 실행과 future 객체 설정을 담당합니다. 마지막으로 future 객체가 반환되는 것을 볼 수 있습니다. 차단하지 않고 바로. 🎜🎜🎜🎜3 .adjust_thread_count 메소드 🎜🎜이 메소드의 의미는 이해하기 쉽습니다. 주로 지정된 수의 스레드를 생성합니다. 그러나 구현이 다소 이해하기 어렵습니다. 예를 들어 스레드 실행 함수의 Weakref.ref에는 약한 참조와 같은 개념이 포함되어 있으며 이에 대해서는 나중에 이해해야 합니다. 🎜🎜🎜🎜4 ._WorkItem 🎜🎜_WorkItem 개체의 책임은 작업을 수행하고 결과를 설정하는 것입니다. 여기서 주요 복잡성은 self.future.set_result(result)🎜입니다. 🎜🎜🎜🎜🎜🎜5. 쓰레드 실행 함수 --_worker🎜🎜 쓰레드 풀이 쓰레드를 생성할 때 지정하는 함수 항목으로 주로 큐에서 🎜task🎜를 꺼내 실행하는데, 기능은 아직 명확하지 않습니다. 나중에 남겨두세요. 🎜🎜🎜🎜

위 내용은 Python ThreadPoolExecutor의 스레드 풀 문제를 해결하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!