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
方法一次取出所有任务的结果。
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循环下面的语句,然后继续阻塞住,循环到所有的任务结束。
从结果也可以看出,先完成的任务会先通知主线程。
除了上面的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
方法可以让主线程阻塞,直到满足设定的要求。
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개의 스레드만 필요합니다. 스레드 생성에는 시스템 리소스 측면에서 더 나은 솔루션이 있습니까?
Python3.2
부터 표준 라이브러리는 ThreadPoolExecutor
및 ProcessPoolExecutorconcurrent.futures
모듈을 제공합니다. /code>두 클래스는 스레딩
및 다중 처리
의 추가 추상화를 실현합니다(여기서는 주로 스레드 풀에 중점을 둡니다). 이 클래스는 스레드를 자동으로 예약하는 데 도움이 될 뿐만 아니라 :🎜 as_completed
메소드를 사용한 결과입니다. 🎜rrreee🎜as_completed()
메서드는 작업이 완료되지 않으면 차단됩니다. 특정 작업이 완료되면 작업이 생성
됩니다. for 루프 아래의 명령문을 실행한 다음 모든 작업이 완료될 때까지 계속 차단할 수 있습니다. 🎜🎜결과에서도 🎜먼저 완료된 작업이 메인 스레드에 먼저 통보됩니다🎜는 것을 알 수 있습니다. 🎜as_completed
메서드 외에 executor.map
메서드를 사용할 수도 있지만 약간의 차이가 있습니다. 🎜rrreee🎜미리 submit
메서드를 사용하지 않고 map
메서드를 사용하세요. map
메서드는 python
과 동일합니다. code> 표준 라이브러리. code>map은 동일한 의미를 가지며 시퀀스의 각 요소에 대해 동일한 기능을 실행합니다. 🎜🎜위 코드는 urls
의 각 요소에 대해 get_html
함수를 실행하고 각 스레드 풀을 할당하는 코드입니다. 위의 as_completed
메소드의 결과와 실행 결과가 다른 것을 확인할 수 있습니다. 🎜urls
🎜목록의 순서와 동일합니다. 2s 작업이 먼저 완료되고, 3s 작업이 먼저 인쇄된 다음 2s 작업이 인쇄됩니다. 🎜wait
메서드는 설정된 요구 사항이 충족될 때까지 메인 스레드를 차단할 수 있습니다. 🎜rrreee🎜 wait
메서드는 대기 작업 순서, 시간 초과 시간, 대기 조건 등 3개의 매개변수를 받습니다. 🎜🎜대기 조건 return_when
은 기본적으로 ALL_COMPLETED
로 설정되어 모든 작업이 끝날 때까지 기다리고 있음을 나타냅니다. 🎜🎜실행 결과에서 모든 작업이 실제로 완료되었음을 확인할 수 있으며, 메인 스레드는 main
을 인쇄합니다. 🎜🎜대기 조건을 FIRST_COMPLETED
로 설정할 수도 있습니다. 즉, 첫 번째 작업이 완료되면 대기가 중지됩니다. 🎜🎜cocurrent.future
모듈의 🎜🎜소스 코드 분석🎜🎜future
은 future 객체를 의미하며, 이는 🎜미래에 완료된 작업🎜으로 이해될 수 있으며, 이는 비동기 기본입니다. 프로그래밍의. 🎜스레드 풀 submit()
이후 future
객체가 반환될 때 작업이 완료되지 않지만 나중에 완료됩니다. submit()
之后,返回的就是这个future
对象,返回的时候任务并没有完成,但会在将来完成。
也可以称之为task的返回容器,这个里面会存储task的结果和状态。
那ThreadPoolExecutor
内部是如何操作这个对象的呢?
下面简单介绍ThreadPoolExecutor
的部分代码:
init
方法中主要重要的就是任务队列和线程集合,在其他方法中需要使用到。
submit
中有两个重要的对象,_base.Future()
和_WorkItem()
对象,_WorkItem()
对象负责运行任务和对future
对象进行设置,最后会将future
对象返回,可以看到整个过程是立即返回的,没有阻塞。
这个方法的含义很好理解,主要是创建指定的线程数。但是实现上有点难以理解,比如线程执行函数中的weakref.ref,涉及到了弱引用等概念,留待以后理解。
_WorkItem
对象的职责就是执行任务和设置结果。这里面主要复杂的还是self.future.set_result(result)
은 task
그렇다면
ThreadPoolExecutor
는 이 개체를 내부적으로 어떻게 작동합니까?
ThreadPoolExecutor
코드 일부에 대한 간략한 소개입니다. 1.init 메소드 init
메소드에서 가장 중요한 것은 작업 대기열입니다. 및 기존의 다른 방법에 필요한 스레드 수집입니다.
submit
, _base.Future()
및 _WorkItem()
객체에는 두 가지 중요한 객체인 _WorkItem이 있습니다. ()
객체는 작업 실행과 future
객체 설정을 담당합니다. 마지막으로 future
객체가 반환되는 것을 볼 수 있습니다. 차단하지 않고 바로. 🎜🎜_WorkItem
개체의 책임은 작업을 수행하고 결과를 설정하는 것입니다. 여기서 주요 복잡성은 self.future.set_result(result)
🎜입니다. 🎜🎜🎜🎜🎜🎜5. 쓰레드 실행 함수 --_worker🎜🎜 쓰레드 풀이 쓰레드를 생성할 때 지정하는 함수 항목으로 주로 큐에서 🎜task🎜를 꺼내 실행하는데, 기능은 아직 명확하지 않습니다. 나중에 남겨두세요. 🎜🎜🎜🎜위 내용은 Python ThreadPoolExecutor의 스레드 풀 문제를 해결하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!