ToB企服应用市场:ToB评测及商务社交产业平台

标题: 利用Python快速实现一个线程池，非常简单 [打印本页]
作者: 杀鸡焉用牛刀    时间: 2022-9-6 01:00
标题: 利用Python快速实现一个线程池，非常简单

雷猴啊，兄弟们！今天来展示一下如何用Python快速实现一个线程池。
 
一、序言

当有多个 IO 密集型的任务要被处理时，我们自然而然会想到多线程。但如果任务非常多，我们不可能每一个任务都启动一个线程去处理，这个时候最好的办法就是实现一个线程池，至于池子里面的线程数量可以根据业务场景进行设置。
比如我们实现一个有 10 个线程的线程池，这样可以并发地处理 10 个任务，每个线程将任务执行完之后，便去执行下一个任务。通过使用线程池，可以避免因线程创建过多而导致资源耗尽，而且任务在执行时的生命周期也可以很好地把控。
而线程池的实现方式也很简单，但这里我们不打算手动实现，因为 Python 提供了一个标准库 concurrent.futures，已经内置了对线程池的支持。所以本篇文章，我们就来详细介绍一下该模块的用法。
二、正文

1、Future 对象

当我们往线程池里面提交一个函数时，会分配一个线程去执行，同时立即返回一个 Future 对象。通过 Future 对象可以监控函数的执行状态，有没有出现异常，以及有没有执行完毕等等。如果函数执行完毕，内部便会调用 future.set_result 将返回值设置到 future 里面，然后外界便可调用 future.result 拿到返回值。
除此之外 future 还可以绑定回调，一旦函数执行完毕，就会以 future 为参数，自动触发回调。所以 future 被称为未来对象，可以把它理解为函数的一个容器，当我们往线程池提交一个函数时，会立即创建相应的 future 然后返回。函数的执行状态什么的，都通过 future 来查看，当然也可以给它绑定一个回调，在函数执行完毕时自动触发。
那么下面我们就来看一下 future 的用法，文字的话理解起来可能有点枯燥。
将函数提交到线程池里面运行时，会立即返回一个对象
这个对象就叫做 Future 对象，里面包含了函数的执行状态等等
当然我们也可以手动创建一个Future对象。
  1. from concurrent.futures import Future
  3. # 创建 Future 对象 future
  4. future = Future()
  6. # 给 future 绑定回调
  7. # Python学习交流群： 279199867
  8. def callback(f: Future):
  9.     print("当set_result的时候会执行回调，result:",
  10.           f.result())
  12. future.add_done_callback(callback)
  13. # 通过 add_done_callback 方法即可给 future 绑定回调
  14. # 调用的时候会自动将 future 作为参数
  15. # 如果需要多个参数，那么就使用偏函数
  17. # 回调函数什么时候执行呢？
  18. # 显然是当 future 执行 set_result 的时候
  19. # 如果 future 是向线程池提交函数时返回的
  20. # 那么当函数执行完毕时会自动执行 future.set_result(xx)
  21. # 并将自身的返回设置进去
  22. # 而这里的 future 是我们手动创建的，因此需要手动执行
  23. future.set_result("嘿嘿")
复制代码
 
 
当set_result的时候会执行回调，result: 嘿嘿
需要注意的是：只能执行一次 set_result，但是可以多次调用 result 获取结果。
  1. from concurrent.futures import Future
  3. future = Future()
  4. future.set_result("哼哼")
  6. print(future.result())  # 哼哼
  7. print(future.result())  # 哼哼
  8. print(future.result())  # 哼哼
复制代码
 
执行 future.result() 之前一定要先 set_result，否则会一直处于阻塞状态。当然 result 方法还可以接收一个 timeout 参数，表示超时时间，如果在指定时间内没有获取到值就会抛出异常。
2、提交函数自动创建 Future 对象

我们上面是手动创建的 Future 对象，但工作中很少会手动创建。我们将函数提交到线程池里面运行的时候，会自动创建 Future 对象并返回。这个 Future 对象里面就包含了函数的执行状态，比如此时是处于暂停、运行中还是完成等等，并且函数在执行完毕之后，还会调用 future.set_result 将自身的返回值设置进去。
  1. from concurrent.futures import ThreadPoolExecutor
  2. import time
  4. def task(name, n):
  5.     time.sleep(n)
  6.     return f"{name} 睡了 {n} 秒"
  8. # 创建一个线程池
  9. # 里面还可以指定 max_workers 参数，表示最多创建多少个线程
  10. # 如果不指定，那么每提交一个函数，都会为其创建一个线程
  11. executor = ThreadPoolExecutor()
  13. # 通过 submit 即可将函数提交到线程池，一旦提交，就会立刻运行
  14. # 因为开启了一个新的线程，主线程会继续往下执行
  15. # 至于 submit 的参数，按照函数名，对应参数提交即可
  16. # 切记不可写成task("古明地觉", 3)，这样就变成调用了
  17. future = executor.submit(task, "屏幕前的你", 3)
  19. # 由于函数里面出现了 time.sleep，并且指定的 n 是 3
  20. # 所以函数内部会休眠 3 秒，显然此时处于运行状态
  21. print(future)
  22. """
  23. <Future at 0x7fbf701726d0 state=running>
  24. """
  26. # 我们说 future 相当于一个容器，包含了内部函数的执行状态
  27. # 函数是否正在运行中
  28. print(future.running())
  29. """
  30. True
  31. """
  32. # 函数是否执行完毕
  33. print(future.done())
  34. """
  35. False
  36. """
  38. # 主程序也 sleep 3 秒
  39. time.sleep(3)
  41. # 显然此时函数已经执行完毕了
  42. # 并且打印结果还告诉我们返回值类型是 str
  43. print(future)
  44. """
  45. <Future at 0x7fbf701726d0 state=finished returned str>
  46. """
  48. print(future.running())
  49. """
  50. False
  51. """
  52. print(future.done())
  53. """
  54. True
  55. """
  57. # 函数执行完毕时，会将返回值设置在 future 里
  58. # 也就是说一旦执行了 future.set_result
  59. # 那么就表示函数执行完毕了，然后外界可以调用 result 拿到返回值
  60. print(future.result())
  61. """
  62. 屏幕前的你 睡了 3 秒
  63. """
复制代码
 
这里再强调一下 future.result()，这一步是会阻塞的，举个例子：
  1. # 提交函数
  2. future = executor.submit(task, "屏幕前的你", 3)
  3. start = time.perf_counter()
  4. future.result()
  5. end = time.perf_counter()
  6. print(end - start)  # 3.00331525
复制代码
 
可以看到，future.result() 这一步花了将近 3s。其实也不难理解，future.result() 是干嘛的？就是为了获取函数的返回值，可函数都还没有执行完毕，它又从哪里获取呢？所以只能先等待函数执行完毕，将返回值通过 set_result 设置到 future 里面之后，外界才能调用 future.result() 获取到值。
如果不想一直等待的话，那么在获取值的时候可以传入一个超时时间。
  1. from concurrent.futures import (
  2.     ThreadPoolExecutor,
  3.     TimeoutError
  4. )
  5. import time
  7. def task(name, n):
  8.     time.sleep(n)
  9.     return f"{name} 睡了 {n} 秒"
  11. executor = ThreadPoolExecutor()
  12. future = executor.submit(task, "屏幕前的你", 3)
  13. try:
  14.     # 1 秒之内获取不到值，抛出 TimeoutError
  15.     res = future.result(1)
  16. except TimeoutError:
  17.     pass
  19. # 再 sleep 2 秒，显然函数执行完毕了
  20. time.sleep(2)
  21. # 获取返回值
  22. print(future.result())
  23. """
  24. 屏幕前的你 睡了 3 秒
  25. """
复制代码
 
当然啦，这么做其实还不够智能，因为我们不知道函数什么时候执行完毕。所以最好的办法还是绑定一个回调，当函数执行完毕时，自动触发回调。
  1. from concurrent.futures import ThreadPoolExecutor
  2. import time
  4. def task(name, n):
  5.     time.sleep(n)
  6.     return f"{name} 睡了 {n} 秒"
  8. def callback(f):
  9.     print(f.result())
  11. executor = ThreadPoolExecutor()
  12. future = executor.submit(task, "屏幕前的你", 3)
  13. # 绑定回调，3 秒之后自动调用
  14. future.add_done_callback(callback)
  15. """
  16. 屏幕前的你 睡了 3 秒
  17. """
复制代码
 
需要注意的是，在调用 submit 方法之后，提交到线程池的函数就已经开始执行了。而不管函数有没有执行完毕，我们都可以给对应的 future 绑定回调。
如果函数完成之前添加回调，那么会在函数完成后触发回调。如果函数完成之后添加回调，由于函数已经完成，代表此时的 future 已经有值了，或者说已经 set_result 了，那么会立即触发回调。
3、future.set_result 到底干了什么事情

当函数执行完毕之后，会执行 set_result，那么这个方法到底干了什么事情呢？

我们看到 future 有两个被保护的属性，分别是 _result 和 _state。显然 _result 用于保存函数的返回值，而 future.result() 本质上也是返回 _result 属性的值。而 _state 属性则用于表示函数的执行状态，初始为 PENDING，执行中为 RUNING，执行完毕时被设置为 FINISHED。
调用 future.result() 的时候，会判断 _state 的属性，如果还在执行中就一直等待。当 _state 为 FINISHED 的时候，就返回 _result 属性的值。
4、提交多个函数

我们上面每次只提交了一个函数，但其实可以提交任意多个，我们来看一下：
  1. from concurrent.futures import ThreadPoolExecutor
  2. import time
  4. def task(name, n):
  5.     time.sleep(n)
  6.     return f"{name} 睡了 {n} 秒"
  8. executor = ThreadPoolExecutor()
  9. futures = [executor.submit(task, "屏幕前的你", 3),
  10.            executor.submit(task, "屏幕前的你", 4),
  11.            executor.submit(task, "屏幕前的你", 1)]
  12. # 此时都处于running
  13. print(futures)
  14. """
  15. [<Future at 0x1b5ff622550 state=running>,
  16. <Future at 0x1b5ff63ca60 state=running>,
  17. <Future at 0x1b5ff63cdf0 state=running>]
  18. """
  20. time.sleep(3)
  21. # 主程序 sleep 3s 后
  22. # futures[0]和futures[2]处于 finished
  23. # futures[1]仍处于 running
  24. print(futures)
  25. """
  26. [<Future at 0x1b5ff622550 state=running>,
  27. <Future at 0x1b5ff63ca60 state=running>,
  28. <Future at 0x1b5ff63cdf0 state=finished returned str>]
  29. """
复制代码
 
如果是多个函数，要如何拿到返回值呢？很简单，遍历 futures 即可。
  1. executor = ThreadPoolExecutor()
  2. futures = [executor.submit(task, "屏幕前的你", 5),
  3.            executor.submit(task, "屏幕前的你", 2),
  4.            executor.submit(task, "屏幕前的你", 4),
  5.            executor.submit(task, "屏幕前的你", 3),
  6.            executor.submit(task, "屏幕前的你", 6)]
  8. for future in futures:
  9.     print(future.result())
  10. """
  11. 屏幕前的你 睡了 5 秒
  12. 屏幕前的你 睡了 2 秒
  13. 屏幕前的你 睡了 4 秒
  14. 屏幕前的你 睡了 3 秒
  15. 屏幕前的你 睡了 6 秒
  16. """
复制代码
 
这里面有一些值得说一说的地方，首先 futures 里面有 5 个 future，记做 future1, future2, future3, future4, future5。
当使用 for 循环遍历的时候，实际上会依次遍历这 5 个 future，所以返回值的顺序就是我们添加的函数的顺序。由于 future1 对应的函数休眠了 5s，那么必须等到 5s 后，future1 里面才会有值。
但这五个函数是并发执行的，future2, future3, future4 由于只休眠了 2s, 4s, 3s，所以肯定会先执行完毕，然后执行 set_result，将返回值设置到对应的 future 里。
但 Python 的 for 循环不可能在第一次迭代还没有结束，就去执行第二次迭代。因为 futures 里面的几个 future 的顺序已经一开始就被定好了，只有当第一个 future.result() 执行完成之后，才会执行第二个 future.result()，以及第三个、第四个。
因此即便后面的函数已经执行完毕，但由于 for 循环的顺序，也只能等着，直到前面的 future.result() 执行完毕。所以当第一个 future.result() 结束时，后面三个 future.result() 会立刻输出，因为它们内部的函数已经执行结束了。
而最后一个 future，由于内部函数 sleep 了 6 秒，因此要再等待 1 秒，才会打印 future.result()。
5、使用 map 来提交多个函数

使用 submit 提交函数会返回一个 future，并且还可以给 future 绑定一个回调。但如果不关心回调的话，那么还可以使用 map 进行提交。
  1. executor = ThreadPoolExecutor()
  2. # map 内部也是使用了 submit
  3. results = executor.map(task,
  4.                        ["屏幕前的你"] * 3,
  5.                        [3, 1, 2])
  6. # 并且返回的是迭代器
  7. print(results)
  8. """
  9. <generator object ... at 0x0000022D78EFA970>
  10. """
  12. # 此时遍历得到的是不再是 future
  13. # 而是 future.result()
  14. for result in results:
  15.     print(result)
  16. """
  17. 屏幕前的你 睡了 3 秒
  18. 屏幕前的你 睡了 1 秒
  19. 屏幕前的你 睡了 2 秒
  20. """
复制代码
 
可以看到，当使用for循环的时候，map 执行的逻辑和 submit 是一样的。唯一的区别是，此时不需要再调用 result 了，因为返回的就是函数的返回值。
或者我们直接调用 list 也行。
  1. executor = ThreadPoolExecutor()
  2. results = executor.map(task,
  3.                        ["屏幕前的你"] * 3,
  4.                        [3, 1, 2])
  5. print(list(results))
  6. """
  7. ['屏幕前的你 睡了 3 秒',
  8. '屏幕前的你 睡了 1 秒',
  9. '屏幕前的你 睡了 2 秒']
  10. """
复制代码
 
results 是一个生成器，调用 list 的时候会将里面的值全部产出。由于 map 内部还是使用的 submit，然后通过 future.result() 拿到返回值，而耗时最长的函数需要 3 秒，因此这一步会阻塞 3 秒。3 秒过后，会打印所有函数的返回值。
6、按照顺序等待执行

上面在获取返回值的时候，是按照函数的提交顺序获取的。如果我希望哪个函数先执行完毕，就先获取哪个函数的返回值，该怎么做呢？
  1. from concurrent.futures import (
  2.     ThreadPoolExecutor,
  3.     as_completed
  4. )
  5. import time
  7. def task(name, n):
  8.     time.sleep(n)
  9.     return f"{name} 睡了 {n} 秒"
  11. executor = ThreadPoolExecutor()
  12. futures = [executor.submit(task, "屏幕前的你", 5),
  13.            executor.submit(task, "屏幕前的你", 2),
  14.            executor.submit(task, "屏幕前的你", 1),
  15.            executor.submit(task, "屏幕前的你", 3),
  16.            executor.submit(task, "屏幕前的你", 4)]
  17. for future in as_completed(futures):
  18.     print(future.result())
  19. """
  20. 屏幕前的你 睡了 1 秒
  21. 屏幕前的你 睡了 2 秒
  22. 屏幕前的你 睡了 3 秒
  23. 屏幕前的你 睡了 4 秒
  24. 屏幕前的你 睡了 5 秒
  25. """
复制代码
 
此时谁先完成，谁先返回。
7、取消一个函数的执行

我们通过 submit 可以将函数提交到线程池中执行，但如果我们想取消该怎么办呢？
  1. executor = ThreadPoolExecutor()
  2. future1 = executor.submit(task, "屏幕前的你", 1)
  3. future2 = executor.submit(task, "屏幕前的你", 2)
  4. future3 = executor.submit(task, "屏幕前的你", 3)
  5. # 取消函数的执行
  6. # 会将 future 的 _state 属性设置为 CANCELLED
  7. future3.cancel()
  8. # 查看是否被取消
  9. print(future3.cancelled())  # False
复制代码
 
问题来了，调用 cancelled 方法的时候，返回的是False，这是为什么？很简单，因为函数已经被提交到线程池里面了，函数已经运行了。而只有在还没有运行时，取消才会成功。
可这不矛盾了吗？函数一旦提交就会运行，只有不运行才会取消成功，这怎么办？还记得线程池的一个叫做 max_workers 的参数吗？用来控制线程池内的线程数量，我们可以将最大的线程数设置为2，那么当第三个函数进去的时候，就不会执行了，而是处于暂停状态。
  1. executor = ThreadPoolExecutor(max_workers=2)
  2. future1 = executor.submit(task, "屏幕前的你", 1)
  3. future2 = executor.submit(task, "屏幕前的你", 2)
  4. future3 = executor.submit(task, "屏幕前的你", 3)
  5. # 如果池子里可以创建空闲线程
  6. # 那么函数一旦提交就会运行，状态为 RUNNING
  7. print(future1._state)  # RUNNING
  8. print(future2._state)  # RUNNING
  9. # 但 future3 内部的函数还没有运行
  10. # 因为池子里无法创建新的空闲线程了，所以状态为 PENDING
  11. print(future3._state)  # PENDING
  12. # 取消函数的执行，前提是函数没有运行
  13. # 会将 future 的 _state 属性设置为 CANCELLED
  14. future3.cancel()
  15. # 查看是否被取消
  16. print(future3.cancelled())  # True
  17. print(future3._state)  # CANCELLED
复制代码
 
在启动线程池的时候，肯定是需要设置容量的，不然处理几千个函数要开启几千个线程吗。另外当函数被取消了，就不可以再调用 future.result() 了，否则的话会抛出 CancelledError。
8、函数执行时出现异常

我们前面的逻辑都是函数正常执行的前提下，但天有不测风云，如果函数执行时出现异常了该怎么办？
  1. from concurrent.futures import ThreadPoolExecutor
  3. def task1():
  4.     1 / 0
  6. def task2():
  7.     pass
  10. executor = ThreadPoolExecutor(max_workers=2)
  11. future1 = executor.submit(task1)
  12. future2 = executor.submit(task2)
  13. print(future1)
  14. print(future2)
  15. """
  16. <Future at 0x7fe3e00f9e50 state=finished raised ZeroDivisionError>
  17. <Future at 0x7fe3e00f9eb0 state=finished returned NoneType>
  18. """
  20. # 结果显示 task1 函数执行出现异常了
  21. # 那么这个异常要怎么获取呢？
  22. print(future1.exception())
  23. print(future1.exception().__class__)
  24. """
  25. division by zero
  26. <class 'ZeroDivisionError'>
  27. """
  29. # 如果执行没有出现异常，那么 exception 方法返回 None
  30. print(future2.exception())  # None
  32. # 注意：如果函数执行出现异常了
  33. # 那么调用 result 方法会将异常抛出来
  34. future1.result()
  35. """
  36. Traceback (most recent call last):
  37.   File "...", line 4, in task1
  38.     1 / 0
  39. ZeroDivisionError: division by zero
  40. """
复制代码
 
出现异常时，调用 future.set_exception 将异常设置到 future 里面，而 future 有一个 _exception 属性，专门保存设置的异常。当调用 future.exception() 时，也会直接返回 _exception 属性的值。
9、等待所有函数执行完毕

假设我们往线程池提交了很多个函数，如果希望提交的函数都执行完毕之后，主程序才能往下执行，该怎么办呢？其实方案有很多：
第一种：
  1. from concurrent.futures import ThreadPoolExecutor
  2. import time
  4. def task(n):
  5.     time.sleep(n)
  6.     return f"sleep {n}"
  8. executor = ThreadPoolExecutor()
  10. future1 = executor.submit(task, 5)
  11. future2 = executor.submit(task, 2)
  12. future3 = executor.submit(task, 4)
  14. # 这里是不会阻塞的
  15. print("start")
  16. # 遍历所有的 future，并调用其 result 方法
  17. # 这样就会等到所有的函数都执行完毕之后才会往下走
  18. for future in [future1, future2, future3]:
  19.     print(future.result())
  20. print("end")
  21. """
  22. start
  23. sleep 5
  24. sleep 2
  25. sleep 4
  26. end
  27. """
复制代码
 
第二种：
  1. from concurrent.futures import (
  2.     ThreadPoolExecutor,
  3.     wait
  4. )
  5. import time
  7. def task(n):
  8.     time.sleep(n)
  9.     return f"sleep {n}"
  11. executor = ThreadPoolExecutor()
  13. future1 = executor.submit(task, 5)
  14. future2 = executor.submit(task, 2)
  15. future3 = executor.submit(task, 4)
  17. # return_when 有三个可选参数
  18. # FIRST_COMPLETED：当任意一个任务完成或者取消
  19. # FIRST_EXCEPTION：当任意一个任务出现异常
  20. #                  如果都没出现异常等同于ALL_COMPLETED
  21. # ALL_COMPLETED：所有任务都完成，默认是这个值
  22. fs = wait([future1, future2, future3],
  23.           return_when="ALL_COMPLETED")
  24. # 此时返回的fs是DoneAndNotDoneFutures类型的namedtuple
  25. # 里面有两个值，一个是done，一个是not_done
  26. print(fs.done)
  27. """
  28. {<Future at 0x1df1400 state=finished returned str>,
  29. <Future at 0x2f08e48 state=finished returned str>,
  30. <Future at 0x9f7bf60 state=finished returned str>}
  31. """
  33. print(fs.not_done)
  34. """
  35. set()
  36. """
  37. for f in fs.done:
  38.     print(f.result())
  39. """
  40. start
  41. sleep 5
  42. sleep 2
  43. sleep 4
  44. end
  45. """
复制代码
 
第三种：
  1. # 使用上下文管理
  2. with ThreadPoolExecutor() as executor:
  3.     future1 = executor.submit(task, 5)
  4.     future2 = executor.submit(task, 2)
  5.     future3 = executor.submit(task, 4)
  7. # 所有函数执行完毕（with语句结束）后才会往下执行
复制代码
 
第四种：
  1. executor = ThreadPoolExecutor()
  3. future1 = executor.submit(task, 5)
  4. future2 = executor.submit(task, 2)
  5. future3 = executor.submit(task, 4)
  6. # 所有函数执行结束后，才会往下执行
  7. executor.shutdown()
复制代码
 
三、小结

如果我们需要启动多线程来执行函数的话，那么不妨使用线程池。每调用一个函数就从池子里面取出一个线程，函数执行完毕就将线程放回到池子里以便其它函数执行。如果池子里面空了，或者说无法创建新的空闲线程，那么接下来的函数就只能处于等待状态了。
最后，concurrent.futures 不仅可以用于实现线程池，还可以用于实现进程池。两者的 API 是一样的：
  1. from concurrent.futures import ProcessPoolExecutor
  2. import time
  4. def task(n):
  5.     time.sleep(n)
  6.     return f"sleep {n}"
  8. executor = ProcessPoolExecutor()
  9. # Windows 上需要加上这一行
  10. if __name__ == '__main__':
  11.     future1 = executor.submit(task, 5)
  12.     future2 = executor.submit(task, 2)
  13.     future3 = executor.submit(task, 4)
  14.     executor.shutdown()
  15.     print(future1.result())
  16.     print(future2.result())
  17.     print(future3.result())
  18. """
  19. sleep 5
  20. sleep 2
  21. sleep 4
  22. """   
复制代码
 
线程池和进程池的 API 是一致的，但工作中很少会创建进程池。
兄弟们今天的分享就到这，债见！
文章不过瘾？试试看视频吧！
Python爬虫入门到实战全集100集教程：代码总是学完就忘记？100个爬虫实战项目！让你沉迷学习丨学以致用丨下一个Python大神就是你！Python tkinter 合集：全网最全python tkinter教程！包含所有知识点！轻松做出好看的tk程序！
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4