Python网络编程之线程与进程

zjp0633

What is a Thread？
　　线程是操作系统能够进行运算调度的最小单位，它被包含在进程之中，是进程中的实际运作单位，一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。
　　在同一个进程内的线程的数据是可以进行互相访问的。
　　线程的切换使用过上下文来实现的，比如有一本书，有a和b这两个人(两个线程)看，a看完之后记录当前看到那一页哪一行，然后交给b看，b看完之后记录当前看到了那一页哪一行，此时a又要看了，那么a就通过上次记录的值(上下文)直接找到上次看到了哪里，然后继续往下看。
What is a Process？
　　一个进程至少要包含一个线程，每个进程在启动的时候就会自动的启动一个线程，进程里面的第一个线程就是主线程，每次在进程内创建的子线程都是由主线程进程创建和销毁，子线程也可以由主线程创建出来的线程创建和销毁线程。
　　进程是对各种资源管理的集合，比如要调用内存、CPU、网卡、声卡等，进程要操作上述的硬件之前都必须要创建一个线程，进程里面可以包含多个线程，QQ就是一个进程。
　　继续拿QQ来说，比如我现在打卡了QQ的聊天窗口、个人信息窗口、设置窗口等，那么每一个打开的窗口都是一个线程，他们都在执行不同的任务，比如聊天窗口这个线程可以和好友进行互动，聊天，视频等，个人信息窗口我可以查看、修改自己的资料。
　　为了进程安全起见，所以两个进程之间的数据是不能够互相访问的(默认情况下)，比如自己写了一个应用程序，然后让别人运行起来，那么我的这个程序就可以访问用户启动的其他应用，我可以通过我自己的程序去访问QQ，然后拿到一些聊天记录等比较隐秘的信息，那么这个时候就不安全了，所以说进程与进程之间的数据是不可以互相访问的，而且每一个进程的内存是独立的。
进程与线程的区别？

• 　　线程是执行的指令集，进程是资源的集合
  
• 　　线程的启动速度要比进程的启动速度要快
  
• 　　两个线程的执行速度是一样的
  
• 　　进程与线程的运行速度是没有可比性的
  
• 　　线程共享创建它的进程的内存空间，进程的内存是独立的。
  
• 　　两个线程共享的数据都是同一份数据，两个子进程的数据不是共享的，而且数据是独立的;
  
• 　　同一个进程的线程之间可以直接交流，同一个主进程的多个子进程之间是不可以进行交流，如果两个进程之间需要通信，就必须要通过一个中间代理来实现;
  
• 　　一个新的线程很容易被创建，一个新的进程创建需要对父进程进行一次克隆
  
• 　　一个线程可以控制和操作同一个进程里的其他线程，线程与线程之间没有隶属关系，但是进程只能操作子进程
  
• 　　改变主线程，有可能会影响到其他线程的行为，但是对于父进程的修改是不会影响子进程;
  

　　一个多并发的小脚本

• 　　import threading
  
• 　　import time
  
• 　　def Princ(String):
  
• 　　print('task', String)
  
• 　　time.sleep(5)
  
• 　　# target=目标函数， args=传入的参数
  
• 　　t1 = threading.Thread(target=Princ, args=('t1',))
  
• 　　t1.start()
  
• 　　t2 = threading.Thread(target=Princ, args=('t1',))
  
• 　　t2.start()
  
• 　　t3 = threading.Thread(target=Princ, args=('t1',))
  
• 　　t3.start()
  

　　参考文档
　　进程与线程的一个简单解释
　　http://www.ruanyifeng.com/blog/2013/04/processes_and_threads.html
　　Linux进程与线程的区别
　　https://my.oschina.net/cnyinlinux/blog/422207
线程
　　Thread module emulating a subset of Java’s threading model.
　　调用threading模块调用线程的两种方式
　　直接调用

• 　　import threading
  
• 　　import time
  
• 　　def Princ(String):
  
• 　　print('task', String)
  
• 　　time.sleep(5)
  
• 　　# target=目标函数， args=传入的参数
  
• 　　t1 = threading.Thread(target=Princ, args=('t1',))
  
• 　　t1.start()
  
• 　　t2 = threading.Thread(target=Princ, args=('t1',))
  
• 　　t2.start()
  
• 　　t3 = threading.Thread(target=Princ, args=('t1',))
  
• 　　t3.start()
  

　　通过类调用

• 　　import threading
  
• 　　import time
  
• 　　class MyThreading(threading.Thread):
  
• 　　def __init__(self, conn):
  
• 　　super(MyThreading, self).__init__()
  
• 　　self.conn = conn
  
• 　　def run(self):
  
• 　　print('run task', self.conn)
  
• 　　time.sleep(5)
  
• 　　t1 = MyThreading('t1')
  
• 　　t2 = MyThreading('t2')
  
• 　　t1.start()
  
• 　　t2.start()
  

多线程
　　多线程在Python内实则就是一个假象，为什么这么说呢，因为CPU的处理速度是很快的，所以我们看起来以一个线程在执行多个任务，每个任务的执行速度是非常之快的，利用上下文切换来快速的切换任务，以至于我们根本感觉不到。
　　但是频繁的使用上下文切换也是要耗费一定的资源，因为单线程在每次切换任务的时候需要保存当前任务的上下文。
　　什么时候用到多线程？
　　首先IO操作是不占用CPU的，只有计算的时候才会占用CPU(譬如1+1=2)，Python中的多线程不适合CPU密集型的任务，适合IO密集型的任务(sockt server)。
　　启动多个线程
　　主进程在启动之后会启动一个主线程，下面的脚本中让主线程启动了多个子线程，然而启动的子线程是独立的，所以主线程不会等待子线程执行完毕，而是主线程继续往下执行，并行执行。

• 　　for i in range(50):
  
• 　　t = threading.Thread(target=Princ, args=('t-%s' % (i),))
  
• 　　t.start()
  

　　join()
　　join()方法可以让程序等待每一个线程之后完成之后再往下执行，又成为串行执行。

• 　　import threading
  
• 　　import time
  
• 　　def Princ(String):
  
• 　　print('task', String)
  
• 　　time.sleep(1)
  
• 　　for i in range(50):
  
• 　　t = threading.Thread(target=Princ, args=('t-%s' % (i),))
  
• 　　t.start()
  
• 　　# 当前线程执行完毕之后在执行后面的线程
  
• 　　t.join()
  

　　让主线程阻塞，子现在并行执行

• 　　import threading
  
• 　　import time
  
• 　　def Princ(String):
  
• 　　print('task', String)
  
• 　　time.sleep(2)
  
• 　　# 执行子线程的时间
  
• 　　start_time = time.time()
  
• 　　# 存放线程的实例
  
• 　　t_objs = []
  
• 　　for i in range(50):
  
• 　　t = threading.Thread(target=Princ, args=('t-%s' % (i),))
  
• 　　t.start()
  
• 　　# 为了不让后面的子线程阻塞，把当前的子线程放入到一个列表中
  
• 　　t_objs.append(t)
  
• 　　# 循环所有子线程实例，等待所有子线程执行完毕
  
• 　　for t in t_objs:
  
• 　　t.join()
  
• 　　# 当前时间减去开始时间就等于执行的过程中需要的时间
  
• 　　print(time.time() - start_time)
  

　　查看主线程与子线程

• 　　import threading
  
• 　　class MyThreading(threading.Thread):
  
• 　　def __init__(self):
  
• 　　super(MyThreading, self).__init__()
  
• 　　def run(self):
  
• 　　print('我是子线程： ', threading.current_thread())
  
• 　　t = MyThreading()
  
• 　　t.start()
  
• 　　print('我是主线程： ', threading.current_thread())
  

　　输出如下：

• 　　C:\Python\Python35\python.exe E:/MyCodeProjects/进程与线程/s3.py
  
• 　　我是子线程：  <MyThreading(Thread-1, started 7724)>
  
• 　　我是主线程：  <_MainThread(MainThread, started 3680)>
  
• 
  
• 　　Process finished with exit code 0
  

　　查看当前进程的活动线程个数

• 　　import threading
  
• 　　class MyThreading(threading.Thread):
  
• 　　def __init__(self):
  
• 　　super(MyThreading, self).__init__()
  
• 　　def run(self):
  
• 　　print('www.anshengme.com')
  
• 　　t = MyThreading()
  
• 　　t.start()
  
• 　　print('线程个数： ', threading.active_count())
  

　　输出如下：

• 　　C:\Python\Python35\python.exe E:/MyCodeProjects/进程与线程/s3.py
  
• 　　www.anshengme.com
  
• 　　# 一个主线程和一个子线程
  
• 　　线程个数：  2
  
• 
  
• 　　Process finished with exit code 0
  

　　Event
　　Event是线程间通信最间的机制之一：一个线程发送一个event信号，其他的线程则等待这个信号。用于主线程控制其他线程的执行。 Events 管理一个flag，这个flag可以使用set
　　()设置成True或者使用clear()重置为False，wait()则用于阻塞，在flag为True之前。flag默认为False。
选项描述Event.wait([timeout])堵塞线程，直到Event对象内部标识位被设为True或超时（如果提供了参数timeout）Event.set()将标识位设为TureEvent.clear()将标识伴设为FalseEvent.isSet()判断标识位是否为Ture

• 　　#!/use/bin/env python
  
• 　　# _*_ coding: utf-8- _*_
  
• 
  
• 　　import threading
  
• 
  
• 　　def runthreading(event):
  
• 　　print("Start...")
  
• 　　event.wait()
  
• 　　print("End...")
  
• 　　event_obj = threading.Event()
  
• 　　for n in range(10):
  
• 　　t = threading.Thread(target=runthreading, args=(event_obj,))
  
• 　　t.start()
  
• 
  
• 　　event_obj.clear()
  
• 　　inp = input("True/False?>> ")
  
• 　　if inp == "True":
  
• 　　event_obj.set()
  
• 　　`
  

　　守护进程(守护线程)
　　一个主进程可以启动多个守护进程，但是主进程必须要一直运行，如果主进程挂掉了，那么守护进程也会随之挂掉
　　程序会等待主线程(进程)执行完毕，但是不会等待守护进程(线程)

• 　　import threading
  
• 　　import time
  
• 
  
• 　　def Princ(String):
  
• 　　print('task', String)
  
• 　　time.sleep(2)
  
• 　　for i in range(50):
  
• 　　t = threading.Thread(target=Princ, args=('t-%s' % (i),))
  
• 　　t.setDaemon(True)  # 把当前线程设置为守护线程，要在start之前设置
  
• 　　t.start()
  

　　场景预设： 比如现在有一个FTP服务，每一个用户连接上去的时候都会创建一个守护线程，现在已经有300个用户连接上去了，就是说已经创建了300个守护线程，但是突然之间FTP服务宕掉了，这个时候就不会等待守护线程执行完毕再退出，而是直接退出，如果是普通的线程，那么就会登台线程执行完毕再退出。

• 　　#!/use/bin/env python
  
• 　　# _*_ coding:utf-8 _*_
  
• 
  
• 　　from multiprocessing import Process
  
• 　　import time
  
• 　　def runprocess(arg):
  
• 　　print(arg)
  
• 　　time.sleep(2)
  
• 
  
• 
  
• 　　p = Process(target=runprocess, args=(11,))
  
• 　　p.daemon=True
  
• 　　p.start()
  
• 
  
• 　　print("end")
  

　　线程之间的数据交互与锁(互斥锁)
　　python2.x需要加锁，但是在python3.x上面就不需要了

• 　　# _*_ coding:utf-8 _*_
  
• 　　import threading
  
• 　　def Princ():
  
• 　　# 获取锁
  
• 　　lock.acquire()
  
• 　　# 在函数内可以直接修改全局变量
  
• 　　global number
  
• 　　number += 1
  
• 　　# 为了避免让程序出现串行，不能加sleep
  
• 　　# time.sleep(1)
  
• 　　# 释放锁
  
• 　　lock.release()
  
• 　　# 锁
  
• 　　lock = threading.Lock()
  
• 　　# 主线程的number
  
• 　　number = 0
  
• 　　t_objs = []
  
• 　　for i in range(100):
  
• 　　t = threading.Thread(target=Princ)
  
• 　　t.start()
  
• 　　t_objs.append(t)
  
• 　　for t in t_objs:
  
• 　　t.join()
  
• 　　print('Number:', number)
  

　　递归锁(Lock/RLock)

• 　　import threading
  
• 　　def run1():
  
• 　　print("grab the first part data")
  
• 　　lock.acquire()
  
• 　　global num
  
• 　　num += 1
  
• 　　lock.release()
  
• 　　return num
  
• 　　def run2():
  
• 　　print("grab the second part data")
  
• 　　lock.acquire()
  
• 　　global num2
  
• 　　num2 += 1
  
• 　　lock.release()
  
• 　　return num2
  
• 　　def run3():
  
• 　　lock.acquire()
  
• 　　res = run1()
  
• 　　print('--------between run1 and run2-----')
  
• 　　res2 = run2()
  
• 　　lock.release()
  
• 　　print(res, res2)
  
• 　　t_objs = []
  
• 　　if __name__ == '__main__':
  
• 　　num, num2 = 0, 0
  
• 　　lock = threading.RLock()  # RLock()类似创建了一个字典，每次退出的时候找到字典的值进行退出
  
• 　　# lock = threading.Lock()  # Lock()会阻塞在这儿
  
• 　　for i in range(10):
  
• 　　t = threading.Thread(target=run3)
  
• 　　t.start()
  
• 　　t_objs.append(t)
  
• 　　for t in t_objs:
  
• 　　t.join()
  
• 　　print(num, num2)
  

　　信号量(Semaphore)
　　互斥锁同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据

• 　　import threading
  
• 　　import time
  
• 　　def run(n):
  
• 　　semaphore.acquire()  # 获取信号，信号可以有多把锁
  
• 　　time.sleep(1)  # 等待一秒钟
  
• 　　print("run the thread: %s\n" % n)
  
• 　　semaphore.release()  # 释放信号
  
• 　　t_objs = []
  
• 　　if __name__ == '__main__':
  
• 　　semaphore = threading.BoundedSemaphore(5)  # 声明一个信号量，最多允许5个线程同时运行
  
• 　　for i in range(20):  # 运行20个线程
  
• 　　t = threading.Thread(target=run, args=(i,))  # 创建线程
  
• 　　t.start()  # 启动线程
  
• 　　t_objs.append(t)
  
• 　　for t in t_objs:
  
• 　　t.join()
  
• 　　print('>>>>>>>>>>>>>')
  

　　以上代码中，类似与创建了一个队列，最多放5个任务，每执行完成一个任务就会往后面增加一个任务。
多进程
　　多进程的资源是独立的，不可以互相访问。
　　启动一个进程

• 　　from multiprocessing import Process
  
• 　　import time
  
• 　　def f(name):
  
• 　　time.sleep(2)
  
• 　　print('hello', name)
  
• 　　if __name__ == '__main__':
  
• 　　# 创建一个进程
  
• 　　p = Process(target=f, args=('bob',))
  
• 　　# 启动
  
• 　　p.start()
  
• 　　# 等待进程执行完毕
  
• 　　p.join(
  

　　在进程内启动一个线程

• 　　from multiprocessing import Process
  
• 　　import threading
  
• 　　def Thread(String):
  
• 　　print(String)
  
• 　　def Proces(String):
  
• 　　print('hello', String)
  
• 　　t = threading.Thread(target=Thread, args=('Thread %s' % (String),))  # 创建一个线程
  
• 　　t.start()  # 启动它
  
• 　　if __name__ == '__main__':
  
• 　　p = Process(target=Proces, args=('World',))  # 创建一个进程
  
• 　　p.start()  # 启动
  
• 　　p.join()  # 等待进程执行完毕
  

　　启动一个多进程

• 　　from multiprocessing import Process
  
• 　　import time
  
• 　　def f(name):
  
• 　　time.sleep(2)
  
• 　　print('hello', name)
  
• 　　if __name__ == '__main__':
  
• 　　for n in range(10):  # 创建一个进程
  
• 　　p = Process(target=f, args=('bob %s' % (n),))
  
• 　　# 启动
  
• 　　p.start()
  
• 　　# 等待进程执行完毕
  

　　获取启动进程的PID

• 　　# _*_ coding:utf-8 _*_
  
• 　　from multiprocessing import Process
  
• 　　import os
  
• 　　def info(String):
  
• 　　print(String)
  
• 　　print('module name:', __name__)
  
• 　　print('父进程的PID:', os.getppid())
  
• 　　print('子进程的PID:', os.getpid())
  
• 　　print("\n")
  
• 　　def ChildProcess():
  
• 　　info('\033[31;1mChildProcess\033[0m')
  
• 　　if __name__ == '__main__':
  
• 　　info('\033[32;1mTheParentProcess\033[0m')
  
• 　　p = Process(target=ChildProcess)
  
• 　　p.start()
  

　　输出结果

• 　　C:\Python\Python35\python.exe E:/MyCodeProjects/多进程/s1.py
  
• 　　TheParentProcess
  
• 　　module name: __main__
  
• 　　# Pycharm的PID
  
• 　　父进程的PID: 6888
  
• 　　# 启动的脚本PID
  
• 　　子进程的PID: 4660
  
• 
  
• 　　ChildProcess
  
• 　　module name: __mp_main__
  
• 　　# 脚本的PID
  
• 　　父进程的PID: 4660
  
• 　　# 父进程启动的子进程PID
  
• 　　子进程的PID: 8452
  
• 
  
• 　　Process finished with exit code 0
  

进程间通信
　　默认情况下进程与进程之间是不可以互相通信的，若要实现互相通信则需要一个中间件，另个进程之间通过中间件来实现通信，下面是进程间通信的几种方式。
　　进程Queue

• 　　# _*_ coding:utf-8 _*_
  
• 　　from multiprocessing import Process, Queue
  
• 　　def ChildProcess(Q):
  
• 　　Q.put(['Hello', None, 'World'])  # 在Queue里面上传一个列表
  
• 　　if __name__ == '__main__':
  
• 　　q = Queue()  # 创建一个Queue
  
• 　　p = Process(target=ChildProcess, args=(q,))  # 创建一个子进程，并把Queue传给子进程,相当于克隆了一份Queue
  
• 　　p.start()  # 启动子进程
  
• 　　print(q.get())  # 获取q中的数据
  
• 　　p.join()
  

　　管道(Pipes)

• 　　# _*_ coding:utf-8 _*_
  
• 　　from multiprocessing import Process, Pipe
  
• 　　def ChildProcess(conn):
  
• 　　conn.send(['Hello', None, 'World'])  # 写一段数据
  
• 　　conn.close()  # 关闭
  
• 　　if __name__ == '__main__':
  
• 　　parent_conn, child_conn = Pipe()  # 生成一个管道实例，parent_conn, child_conn管道的两头
  
• 　　p = Process(target=ChildProcess, args=(child_conn,))
  
• 　　p.start()
  
• 　　print(parent_conn.recv())  # 收取消息
  
• 　　p.join()
  

　　数据共享(Managers)

• 　　# _*_ coding:utf-8 _*_
  
• 　　# _*_ coding:utf-8 _*_
  
• 　　from multiprocessing import Process, Manager
  
• 　　import os
  
• 
  
• 　　def ChildProcess(Dict, List):
  
• 　　Dict['k1'] = 'v1'
  
• 　　Dict['k2'] = 'v2'
  
• 　　List.append(os.getpid())  # 获取子进程的PID
  
• 　　print(List)  # 输出列表中的内容
  
• 
  
• 　　if __name__ == '__main__':
  
• 　　manager = Manager()  # 生成Manager对象
  
• 　　Dict = manager.dict()  # 生成一个可以在多个进程之间传递共享的字典
  
• 　　List = manager.list()  # 生成一个字典
  
• 
  
• 　　ProcessList = []  # 创建一个空列表，存放进程的对象，等待子进程执行用于
  
• 
  
• 　　for i in range(10):  # 生成是个子进程
  
• 　　p = Process(target=ChildProcess, args=(Dict, List))  # 创建一个子进程
  
• 　　p.start()  # 启动
  
• 　　ProcessList.append(p)  # 把子进程添加到p_list列表中
  
• 
  
• 　　for res in ProcessList:  # 循环所有的子进程
  
• 　　res.join()  # 等待执行完毕
  
• 　　print('\n')
  
• 　　print(Dict)
  
• 　　print(List)
  

　　输出结果

• 　　C:\Python\Python35\python.exe E:/MyCodeProjects/多进程/s4.py
  
• 　　[5112]
  
• 　　[5112, 3448]
  
• 　　[5112, 3448, 4584]
  
• 　　[5112, 3448, 4584, 2128]
  
• 　　[5112, 3448, 4584, 2128, 11124]
  
• 　　[5112, 3448, 4584, 2128, 11124, 10628]
  
• 　　[5112, 3448, 4584, 2128, 11124, 10628, 5512]
  
• 　　[5112, 3448, 4584, 2128, 11124, 10628, 5512, 10460]
  
• 　　[5112, 3448, 4584, 2128, 11124, 10628, 5512, 10460, 10484]
  
• 　　[5112, 3448, 4584, 2128, 11124, 10628, 5512, 10460, 10484, 6804]
  
• 
  
• 
  
• 　　{'k1': 'v1', 'k2': 'v2'}
  
• 　　[5112, 3448, 4584, 2128, 11124, 10628, 5512, 10460, 10484, 6804]
  
• 
  
• 　　Process finished with exit code 0
  

　　锁(Lock)

• 　　from multiprocessing import Process, Lock
  
• 
  
• 　　def ChildProcess(l, i):
  
• 　　l.acquire()  # 获取锁
  
• 　　print('hello world', i)
  
• 　　l.release()  # 释放锁
  
• 
  
• 　　if __name__ == '__main__':
  
• 　　lock = Lock()  # 生成Lock对象
  
• 　　for num in range(10):
  
• 　　Process(target=ChildProcess, args=(lock, num)).start()  # 创建并启动一个子进程
  

进程池
　　同一时间启动多少个进程

• 　　#!/use/bin/env python
  
• 　　# _*_ coding: utf-8 _*_
  
• 
  
• 　　from multiprocessing import Pool
  
• 　　import time
  
• 
  
• 　　def myFun(i):
  
• 　　time.sleep(2)
  
• 　　return i+100
  
• 
  
• 　　def end_call(arg):
  
• 　　print("end_call>>", arg)
  
• 
  
• 　　p = Pool(5)  # 允许进程池内同时放入5个进程
  
• 　　for i in range(10):
  
• 　　p.apply_async(func=myFun, args=(i,),callback=end_call) # # 平行执行,callback是主进程来调用
  
• 　　# p.apply(func=Foo)  # 串行执行
  
• 
  
• 　　print("end")
  
• 　　p.close()
  
• 　　p.join() # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。
  

线程池
　　简单实现

• 　　#!/usr/bin/env python
  
• 　　# -*- coding:utf-8 -*-
  
• 　　import threading
  
• 　　import queue
  
• 　　import time
  
• 　　class MyThread:
  
• 　　def __init__(self,max_num=10):
  
• 　　self.queue = queue.Queue()
  
• 　　for n in range(max_num):
  
• 　　self.queue.put(threading.Thread)
  
• 　　def get_thread(self):
  
• 　　return self.queue.get()
  
• 　　def put_thread(self):
  
• 　　self.queue.put(threading.Thread)
  
• 　　pool = MyThread(5)
  
• 　　def RunThread(arg,pool):
  
• 　　print(arg)
  
• 　　time.sleep(2)
  
• 　　pool.put_thread()
  
• 　　for n in range(30):
  
• 　　thread = pool.get_thread()
  
• 　　t = thread(target=RunThread, args=(n,pool,))
  
• 　　t.start()
  

　　复杂版本

• 　　#!/usr/bin/env python
  
• 　　# -*- coding:utf-8 -*-
  
• 
  
• 　　import queue
  
• 　　import threading
  
• 　　import contextlib
  
• 　　import time
  
• 
  
• 　　StopEvent = object()
  
• 
  
• 　　class ThreadPool(object):
  
• 
  
• 　　def __init__(self, max_num, max_task_num = None):
  
• 　　if max_task_num:
  
• 　　self.q = queue.Queue(max_task_num)
  
• 　　else:
  
• 　　self.q = queue.Queue()
  
• 　　self.max_num = max_num
  
• 　　self.cancel = False
  
• 　　self.terminal = False
  
• 　　self.generate_list = []
  
• 　　self.free_list = []
  
• 
  
• 　　def run(self, func, args, callback=None):
  
• 　　"""
  
• 　　线程池执行一个任务
  
• 　　:param func: 任务函数
  
• 　　:param args: 任务函数所需参数
  
• 　　:param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数1、任务函数执行状态；2、任务函数返回值（默认为None，即：不执行回调函数）
  
• 　　:return: 如果线程池已经终止，则返回True否则None
  
• 　　"""
  
• 　　if self.cancel:
  
• 
  
• 　　return
  
• 　　if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
  
• 　　self.generate_thread()
  
• 　　w = (func, args, callback,)
  
• 　　self.q.put(w)
  
• 
  
• 　　def generate_thread(self):
  
• 　　"""
  
• 　　创建一个线程
  
• 　　"""
  
• 　　t = threading.Thread(target=self.call)
  
• 　　t.start()
  
• 
  
• 　　def call(self):
  
• 　　"""
  
• 　　循环去获取任务函数并执行任务函数
  
• 　　"""
  
• 　　current_thread = threading.currentThread()
  
• 　　self.generate_list.append(current_thread)
  
• 
  
• 　　event = self.q.get()
  
• 　　while event != StopEvent:
  
• 
  
• 　　func, arguments, callback = event
  
• 　　try:
  
• 　　result = func(*arguments)
  
• 　　success = True
  
• 　　except Exception as e:
  
• 　　success = False
  
• 　　result = None
  
• 
  
• 　　if callback is not None:
  
• 　　try:
  
• 　　callback(success, result)
  
• 　　except Exception as e:
  
• 　　pass
  
• 
  
• 　　with self.worker_state(self.free_list, current_thread):
  
• 　　if self.terminal:
  
• 　　event = StopEvent
  
• 　　else:
  
• 　　event = self.q.get()
  
• 　　else:
  
• 
  
• 　　self.generate_list.remove(current_thread)
  
• 
  
• 　　def close(self):
  
• 　　"""
  
• 　　执行完所有的任务后，所有线程停止
  
• 　　"""
  
• 　　self.cancel = True
  
• 　　full_size = len(self.generate_list)
  
• 　　while full_size:
  
• 　　self.q.put(StopEvent)
  
• 　　full_size -= 1
  
• 
  
• 　　def terminate(self):
  
• 　　"""
  
• 　　无论是否还有任务，终止线程
  
• 　　"""
  
• 　　self.terminal = True
  
• 
  
• 　　while self.generate_list:
  
• 　　self.q.put(StopEvent)
  
• 
  
• 　　self.q.queue.clear()
  
• 
  
• 　　@contextlib.contextmanager
  
• 　　def worker_state(self, state_list, worker_thread):
  
• 　　"""
  
• 　　用于记录线程中正在等待的线程数
  
• 　　"""
  
• 　　state_list.append(worker_thread)
  
• 　　try:
  
• 　　yield
  
• 　　finally:
  
• 　　state_list.remove(worker_thread)
  
• 
  
• 　　# How to use
  
• 
  
• 　　pool = ThreadPool(5)
  
• 
  
• 　　def callback(status, result):
  
• 　　# status, execute action status
  
• 　　# result, execute action return value
  
• 　　pass
  
• 
  
• 　　def action(i):
  
• 　　print(i)
  
• 
  
• 　　for i in range(30):
  
• 　　ret = pool.run(action, (i,), callback)
  
• 
  
• 　　time.sleep(5)
  
• 　　print(len(pool.generate_list), len(pool.free_list))
  
• 　　print(len(pool.generate_list), len(pool.free_list))
  
• 　　pool.close()
  
• 　　pool.terminate()
  

什么是IO密集型和CPU密集型？
　　IO密集型(I/O bound)
　　频繁网络传输、读取硬盘及其他IO设备称之为IO密集型，最简单的就是硬盘存取数据，IO操作并不会涉及到CPU。
　　计算密集型(CPU bound)
　　程序大部分在做计算、逻辑判断、循环导致cpu占用率很高的情况，称之为计算密集型，比如说python程序中执行了一段代码1+1，这就是在计算1+1的值