线程池

Posted 2021-01-08 qq752059037

概述

传统多线程方案会使用“即时创建， 即时销毁”的策略。尽管与创建进程相比，创建线程的时间已经大大的缩短，但是如果提交给线程的任务是执行时间较短，而且执行次数极其频繁，那么服务器将处于不停的创建线程，销毁线程的状态。

一个线程的运行时间可以分为3部分：线程的启动时间、线程体的运行时间和线程的销毁时间。在多线程处理的情景中，如果线程不能被重用，就意味着每次创建都需要经过启动、销毁和运行3个过程。这必然会增加系统相应的时间，降低了效率。

线程池的优点

由于线程预先被创建并放入线程池中，同时处理完当前任务之后并不销毁而是被安排处理下一个任务，因此能够避免多次创建线程，从而节省线程创建和销毁的开销，能带来更好的性能和系统稳定性。

线程池原理

线程池基本原理： 我们把任务放进队列中去，然后开N个线程，每个线程都去队列中取一个任务，执行完了之后告诉系统说我执行完了，然后接着去队列中取下一个任务，直至队列中所有任务取空，退出线程。

步骤:

1. 创建Queue.Queue()实例,然后put数据或任务
2. 生成守护线程池,把线程设置成了daemon守护线程
3. 每个线程无限循环阻塞读取queue队列的项目,并处理
4. 每次完成一次工作后,使用queue.task_done()函数向任务已经完成的队列发送一个信号
5. 主线程设置queue.join()阻塞,直到任务队列已经清空了,解除阻塞,向下执行

 

这个模式下有几个注意的点：

• 将线程池的线程设置成daemon守护进程，意味着主线程退出时，守护线程也会自动退出，如果使用默认
  daemon=False的话， 非daemon的线程会阻塞主线程的退出，所以即使queue队列的任务已经完成
  线程池依然阻塞无限循环等待任务，使得主线程也不会退出。

• 当主线程使用了queue.join()的时候，说明主线程会阻塞直到queue已经是清空的，而主线程怎么知道queue已经是清空的呢？就是通过每次线程queue.get()后并处理任务后，发送queue.task_done()信号，queue的数据就会减1，直到queue的数据是空的，queue.join()解除阻塞，向下执行。

• 这个模式主要是以队列queue的任务来做主导的，做完任务就退出，由于线程池是daemon的，所以主退出线程池所有线程都会退出。 有别于我们平时可能以队列主导thread.join()阻塞，这种线程完成之前阻塞主线程。看需求使用哪个join()

 

如果是想做完一定数量任务的队列就结束，使用queue.join()，比如爬取指定数量的网页

如果是想线程做完任务就结束，使用thread.join()

 

 

import queue
import time
import threading

def do_job():
    while True:
        i = queue.get()
        time.sleep(1)
        print(‘index %s, curent: %s‘ % (i,threading.current_thread()))
        queue.task_done()
if __name__ == ‘__main__‘:
    queue = queue.Queue()  # 创建队列实例， 用于存储任务
    for i in range(3):# 创建3个线程
        t = threading.Thread(target=do_job)
        t.daemon=True # 设置守护线程 主线程退出，守护线程也会退出，即使正在运行
        t.start()
    time.sleep(1)
    for i in range(10): #塞进10个任务到队列
        queue.put(i)
    queue.join()

使用队列模拟线程池

 创建线程池

 ThreadPoolExecutor

import time
from concurrent.futures import ThreadPoolExecutor

def func(i):
    print(‘进程%s‘%i)
    time.sleep(1)
    print(‘进程%s结束‘%i)

if __name__ == ‘__main__‘:
    tp=ThreadPoolExecutor(5)
    tp.submit(func,1)
    tp.shutdown()
    print(‘主进程结束‘)

开启多个

import time
from concurrent.futures import ThreadPoolExecutor

def func(i):
    print(‘thread‘,i)
    time.sleep(1)
    print(‘thread‘,i,‘end‘)


if __name__ == ‘__main__‘:
    tp=ThreadPoolExecutor(5)
    for i in range(20):
        tp.submit(func,i)
    tp.shutdown()
    print(‘主线程结束‘)

 

线程池要设置为多少？

服务器CPU核数有限，能够同时并发的线程数有限，并不是开得越多越好，以及线程切换是有开销的，如果线程切换过于频繁，反而会使性能降低

线程执行过程中，计算时间分为两部分：

• CPU计算，占用CPU
• 不需要CPU计算，不占用CPU，等待IO返回，比如recv(), accept(), sleep()等操作

那么如果计算时间占50%， 等待时间50%，那么为了利用率达到最高，可以开2个线程：
假如工作时间是2秒， CPU计算完1秒后，线程等待IO的时候需要1秒，此时CPU空闲了，这时就可以切换到另外一个线程，让CPU工作1秒后，线程等待IO需要1秒，此时CPU又可以切回去，第一个线程这时刚好完成了1秒的IO等待，可以让CPU继续工作，就这样循环的在两个线程之前切换操作。

那么如果计算时间占20%， 等待时间80%，那么为了利用率达到最高，可以开5个线程：
可以想象成完成任务需要5秒，CPU占用1秒，等待时间4秒，CPU在线程等待时，可以同时再激活4个线程，这样就把CPU和IO等待时间，最大化的重叠起来

抽象一下，计算线程数设置的公式就是：
N核服务器，通过执行业务的单线程分析出本地计算时间为x，等待时间为y，则工作线程数（线程池线程数）设置为 N*(x+y)/x，能让CPU的利用率最大化。
由于有GIL的影响，python只能使用到1个核，所以这里设置N=1

 

参考

http://python.jobbole.com/87743/
https://www.ibm.com/developerworks/cn/aix/library/au-threadingpython/
http://www.cnblogs.com/goodhacker/p/3359985.html
https://mp.weixin.qq.com/s/BRpngTEFHjzpGv8tkdqmPQ
http://python3-cookbook.readthedocs.io/zh_CN/latest/c12/p07_creating_thread_pool.html
https://www.zhihu.com/question/23474039
https://blog.csdn.net/a519640026/article/details/76157999?utm_source=copy