一、threading模块介绍
multiprocess模块的完全模仿了threading模块的接口，二者在使用层面，有很大的相似性，因而不再详细介绍
官网链接：点击进入
二、开启线程的两种方式
multiprocess模块的完全模仿了threading模块的接口，二者在使用层面，有很大的相似性
import time, random # from multiprocessing import Process from threading import Thread def piao(name): print('%s piaoing' % name) time.sleep(random.randrange(1, 5)) print('%s piao end' % name) if __name__ == '__main__': t1 = Thread(target=piao, args=('egon', )) t1.start() # 主线程向操作系统发信号，又开了一个线程 print("主线程") # 执行角度看是主线程，从资源角度看是主进程 # 这个程序总体是一个进程、两个线程 """ egon piaoing 主进程 egon piao end """
import time, random # from multiprocessing import Process from threading import Thread class MyThread(Thread): def __init__(self, name): super().__init__() self.name = name def run(self): print("%s piaoing" % self.name) time.sleep(random.randrange(1, 5)) print("%s piao end" % self.name) if __name__ == '__main__': t1 = MyThread('egon') t1.start() # 主线程向操作系统发信号，又开了一个线程 print("主线程") """ egon piaoing 主线程 egon piao end """
三、在一个进程下开启线程与在一个进程下开启多个子进程的区别
import time from multiprocessing import Process from threading import Thread def piao(name): print('%s piaoing' % name) time.sleep(2) print('%s piao end' % name) if __name__ == '__main__': # p1 = Process(target=piao, args=('进程', )) # p1.start() """ 主线程 进程 piaoing 进程 piao end """ t1 = Thread(target=piao, args=('线程', )) t1.start() """ 线程 piaoing 主线程 线程 piao end """ print("主线程") # 对比可知，线程开销远小于进程，因为进程需要申请内存空间。
from threading import Thread from multiprocessing import Process n = 100 def task(): global n n = 0 if __name__ == '__main__': """进程验证： p1 = Process(target=task,) p1.start() # 会把子进程的n改为了0，看是否影响主进程 p1.join() print("主进程", n) # 主进程 100 # 由此可见进程间是隔离的，子进程变量修改不影响主进程 """ """线程验证""" t1 = Thread(target=task, ) t1.start() t1.join() print("主线程", n) # 主线程 0
from threading import Thread from multiprocessing import Process, current_process # current_process查看进程ID号 import os # os.getpid()也可以查看进程ID n = 100 def task(): # print(current_process().pid) print('子进程PID:%s 父进程的PID：%s' % (os.getpid(), os.getppid())) if __name__ == '__main__': p1 = Process(target=task,) p1.start() # print("主线程", current_process().pid) print("主线程", os.getpid()) """ 主线程 6455 子进程PID:6456 父进程的PID：6455 """
from threading import Thread import os # os.getpid()也可以查看进程ID n = 100 def task(): # print(current_process().pid) print('线程的进程 PID:%s' % os.getpid()) if __name__ == '__main__': t1 = Thread(target=task,) t1.start() # print("主线程", current_process().pid) print("主线程", os.getpid()) """说明两个线程是同一个进程： 线程的进程 PID:6493 主线程 6493 """
四、练习
1、基于多线程实现并发的套接字通信
# -*- coding:utf-8 -*- __author__ = 'Qiushi Huang' from socket import * from threading import Thread # 通讯和建立链接分开，启动不同的线程，大家是并发执行。 def communicate(conn): while True: try: data = conn.recv(1024) if not data:break conn.send(data.upper()) except ConnectionResetError: break conn.close() def server(ip, port): server = socket(AF_INET, SOCK_STREAM) server.bind((ip, port)) server.listen(5) while True: conn, addr = server.accept() # 建链接 t = Thread(target=communicate, args=(conn,)) # 建一个链接创一个线程 t.start() # communicate(conn) server.close() if __name__ == '__main__': server('127.0.0.1', 8091) # 主线程 """ 这种解决方案的问题是：当客户端越来越多后，线程也会越来越多，会带来服务崩溃的问题。 """
# -*- coding:utf-8 -*- __author__ = 'Qiushi Huang' # 使用时，可以一个程序运行多次，这是多个不同的in from socket import * client = socket(AF_INET, SOCK_STREAM) client.connect(("127.0.0.1", 8091)) while True: msg = input(">>").strip() if not msg:continue client.send(msg.encode("utf-8")) data = client.recv(1024) print(data.decode("utf-8")) client.close()
2、编写一个简单的文本处理工具，具备三个任务，一个接收用户输入，一个将用户输入的内容格式化成大写，一个将格式化后的结果存入文件
from threading import Thread msg_l=[] format_l=[] def talk(): while True: msg=input('>>: ').strip() if not msg:continue msg_l.append(msg) def format_msg(): while True: if msg_l: res=msg_l.pop() format_l.append(res.upper()) def save(): while True: if format_l: with open('db.txt','a',encoding='utf-8') as f: res=format_l.pop() f.write('%s ' %res) if __name__ == '__main__': t1=Thread(target=talk) t2=Thread(target=format_msg) t3=Thread(target=save) t1.start() t2.start() t3.start()
五、线程对象的属性和方法
1、Thread实例对象的方法
isAlive(): 返回线程是否活动的。
getName(): 返回线程名。
setName(): 设置线程名。
2、threading模块提供的一些方法
threading.currentThread(): 返回当前的线程变量。
threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
3、属性和方法的应用与验证
from threading import Thread, currentThread # 得到线程对象的方法 from threading import active_count # 得到活跃进程数 from threading import enumerate # 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 import time # 需要注意的是线程没有子线程的概念，线程都是属于进程的 def task(): print("%s is running" % currentThread().getName()) # 对象下有一个getName()方法 time.sleep(2) print("%s is done" % currentThread().getName()) if __name__ == '__main__': getName()方法返回线程名 t = Thread(target=task, name='子线程1') t.start() print("主进程", currentThread().getName()) """ 子线程1 is running 主进程 MainThread 子线程1 is done """
from threading import Thread, currentThread # 得到线程对象的方法 from threading import active_count # 得到活跃进程数 from threading import enumerate # 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 import time def task(): print("%s is running" % currentThread().getName()) # 对象下有一个getName()方法 time.sleep(2) print("%s is done" % currentThread().getName()) if __name__ == '__main__': setName()方法设置线程名 t = Thread(target=task, name='子线程1') t.start() t.setName('儿子线程1') # 修改进程名称 currentThread().setName("主线程") # 设置主线程名称（默认是MainThread） print(t.isAlive()) # 判断线程是否存活 print("主进程", currentThread().getName()) """ 子线程1 is running True 主进程 主线程 儿子线程1 is done """
from threading import Thread, currentThread # 得到线程对象的方法 from threading import active_count # 得到活跃进程数 from threading import enumerate # 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 import time def task(): print("%s is running" % currentThread().getName()) # 对象下有一个getName()方法 time.sleep(2) print("%s is done" % currentThread().getName()) if __name__ == '__main__': t = Thread(target=task, name='子线程1') t.start() t.setName('儿子线程1') # 修改进程名称 t.join() # 主线程等子进程运行完毕再执行 currentThread().setName("主线程") # 设置主线程名称（默认是MainThread） print(t.isAlive()) # 判断线程是否存活 print("主进程", currentThread().getName()) """ 子线程1 is running 儿子线程1 is done False 主进程 主线程 """
from threading import Thread, currentThread # 得到线程对象的方法 from threading import active_count # 得到活跃进程数 from threading import enumerate # 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 import time def task(): print("%s is running" % currentThread().getName()) # 对象下有一个getName()方法 time.sleep(2) print("%s is done" % currentThread().getName()) if __name__ == '__main__': # 测试threading.active_count方法 t = Thread(target=task, name='子线程1') t.start() print(active_count()) """ 子线程1 is running 2 子线程1 is done """
from threading import Thread, currentThread # 得到线程对象的方法 from threading import active_count # 得到活跃进程数 from threading import enumerate # 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 import time def task(): print("%s is running" % currentThread().getName()) # 对象下有一个getName()方法 time.sleep(2) print("%s is done" % currentThread().getName()) if __name__ == '__main__': # 对上面改写添加一个join() t = Thread(target=task, name='子线程1') t.start() t.join() # 运行完才执行主线程，因此后面打印的活跃线程数是一个 print(active_count()) """ 子线程1 is running 子线程1 is done 1 """
from threading import Thread, currentThread # 得到线程对象的方法 from threading import active_count # 得到活跃进程数 from threading import enumerate # 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 import time def task(): print("%s is running" % currentThread().getName()) # 对象下有一个getName()方法 time.sleep(2) print("%s is done" % currentThread().getName()) if __name__ == '__main__': # threading.enumerate()方法：返回一个包含正在运行的线程的list t = Thread(target=task, name='子线程1') t.start() print(enumerate()) """ 子线程1 is running [<_MainThread(MainThread, started 4320744256)>, <Thread(子线程1, started 123145383735296)>] 子线程1 is done """
六、守护线程
一个进程内，如果不开线程，默认就是一个主线程，主线程代码运行完毕，进程被销毁。
一个进程内，开多个线程的情况下，主线程在代码运行完毕后，还要等其他线程工作完才死掉，进程销毁。
守护线程守护主线程，等到主线程死了才会被销毁。在有其他线程的情况下，主线程代码运行完后，等其他非守护线程结束，守护线程才会死掉。
无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁。
需要强调的是：运行完毕并非终止运行。运行完毕的真正含义：
1、对主进程来说，运行完毕指的是主进程代码运行完毕。
2、对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才能运行完毕。
详细解释
1、主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束
2、主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。
from threading import Thread import time def sayhi(name): time.sleep(2) print("%s say hello" % name) if __name__ == '__main__': t = Thread(target=sayhi, args=('egon',)) # 守护线程必须在t.start()前设置 # 守护线程设置方式一： t.daemon=True # 守护线程设置方式二： # t.setDaemon(True) t.start() # 立马创建子线程，但需要等待两秒，因此程序会先执行下面的代码 print("主线程") print(t.is_alive()) # 这一行代码执行完后，主线程执行完毕，由于主线程之外，只有一个守护线程，主线程不需要等守护线程执行结束，因此主线程和守护进程终止，进程结束。 """ 主线程 True """
练习：思考下述代码的执行结果有可能是哪些情况？为什么？
from threading import Thread import time def foo(): print(123) time.sleep(1) print("end123") def bar(): print(456) time.sleep(3) print("end456") if __name__ == '__main__': t1=Thread(target=foo) t2=Thread(target=bar) t1.daemon=True # t1是守护线程 t1.start() t2.start() print("main-------") # 主线程结束后，会等待非守护线程结束 # 由于非守护线程需要等待的时间比守护线程长，因此线程都会得到执行 """ 123 456 main------ end123 end456 """
七、GIL全局解释锁(Global Interpreter Lock)
链接：http://www.cnblogs.com/linhaifeng/articles/7449853.html
后期需要详细分析这个部分的内容。
八、同步锁
1、三个需要注意的点
1、线程抢的是GIL锁，GIL锁相当于执行权限，拿到执行权限后才能拿到互斥锁Lock，其他线程也可以抢到GIL，但如果发现Lock仍然没有被释放则阻塞，即便是拿到执行权限GIL也要立刻交出来。
2、join是等待所有，即整体串行，而锁只是锁住修改共享数据的部分，即部分串行，要想保证数据安全的根本原理在于让并发变成串行，join与互斥锁都可以实现，毫无疑问，互斥锁的部分串行效率要更高。
3、一定要主要本小节最后GIL和互斥锁的经典分析。
2、GIL和Lock的对比
Python已经有了一个GIL来保证同一时间只能有一个线程来执行，为什么还需要lock？
锁的目的是为了保护共享的数据，同一时间只能有一个线程来修改共享的数据。
保护不同的数据就应该加不同的锁。
GIL 与Lock是两把锁，保护的数据不一样，GIL是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），Lock是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理，即Lock。
过程分析：所有线程抢的是GIL锁，或者说所有线程抢的是执行权限
线程1抢到GIL锁，拿到执行权限，开始执行，然后加了一把Lock，还没有执行完毕，即线程1还未释放Lock，有可能线程2抢到GIL锁，开始执行，执行过程中发现Lock还没有被线程1释放，于是线程2进入阻塞，被夺走执行权限，有可能线程1拿到GIL，然后正常执行到释放Lock。。。这就导致了串行运行的效果
既然是串行，那我们执行
t1.start()
t1.join
t2.start()
t2.join()
这也是串行执行啊。
需知join是等待t1所有的代码执行完，相当于锁住了t1的所有代码，而Lock只是锁住一部分操作共享数据的代码。
因为Python解释器帮你自动定期进行内存回收，你可以理解为python解释器里有一个独立的线程，每过一段时间它起wake up做一次全局轮询看看哪些内存数据是可以被清空的，此时你自己的程序 里的线程和 py解释器自己的线程是并发运行的，假设你的线程删除了一个变量，py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了，结果就有可能新赋值的数据被删除了，为了解决类似的问题，python解释器简单粗暴的加了锁，即当一个线程运行时，其它人都不能动，这样就解决了上述的问题， 这可以说是Python早期版本的遗留问题。
锁通常被用来实现对共享资源的同步访问。为每一个共享资源创建一个Lock对象，当你需要访问该资源时，调用acquire方法来获取锁对象（如果其它线程已经获得了该锁，则当前线程需等待其被释放），待资源访问完后，再调用release方法释放锁：
import threading R=threading.Lock() R.acquire() ''' 对公共数据的操作 ''' R.release()
from threading import Thread import os,time def work(): global n temp=n time.sleep(0.1) n=temp-1 if __name__ == '__main__': n=100 l=[] for i in range(100): p=Thread(target=work) l.append(p) p.start() for p in l: p.join() print(n) #结果可能为99
from threading import Thread,Lock import os,time def work(): global n lock.acquire() temp=n time.sleep(0.1) n=temp-1 lock.release() if __name__ == '__main__': lock=Lock() n=100 l=[] for i in range(100): p=Thread(target=work) l.append(p) p.start() for p in l: p.join() print(n) #结果肯定为0，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全
2、GIL锁和互斥锁综合分析
1、100个线程去抢GIL锁，即抢执行权限。
2、肯定有一个线程先抢到GIL（暂且称为线程1），然后开始执行，一旦执行就会拿到lock.acquire()
3、极有可能线程1还未运行完毕，就有另外一个线程2抢到GIL，然后开始运行，但线程2发现互斥锁lock还没有被线程1释放，于是阻塞，被迫交出执行权限，即释放GIL。
4、直到线程1重新抢到GIL，开始从上次暂停的位置继续执行，直到正常释放互斥锁lock，然后其他线程再重复234的过程。
3、互斥锁和join的区别
#不加锁:并发执行,速度快,数据不安全 from threading import current_thread,Thread,Lock import os,time def task(): global n print('%s is running' %current_thread().getName()) temp=n time.sleep(0.5) n=temp-1 if __name__ == '__main__': n=100 lock=Lock() threads=[] start_time=time.time() for i in range(100): t=Thread(target=task) threads.append(t) t.start() for t in threads: t.join() stop_time=time.time() print('主:%s n:%s' %(stop_time-start_time,n)) ''' Thread-1 is running Thread-2 is running ...... Thread-100 is running 主:0.5216062068939209 n:99 '''
#加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全 from threading import current_thread,Thread,Lock import os,time def task(): #未加锁的代码并发运行 time.sleep(3) print('%s start to run' %current_thread().getName()) global n #加锁的代码串行运行 lock.acquire() temp=n time.sleep(0.5) n=temp-1 lock.release() if __name__ == '__main__': n=100 lock=Lock() threads=[] start_time=time.time() for i in range(100): t=Thread(target=task) threads.append(t) t.start() for t in threads: t.join() stop_time=time.time() print('主:%s n:%s' %(stop_time-start_time,n)) ''' Thread-1 is running Thread-2 is running ...... Thread-100 is running 主:53.294203758239746 n:0 '''
加锁会将运行变成串行，同样适用join也可以得到串行的效果，数据也是安全的，但是start后立即join，任务内的所有代码都是串行执行的，而加锁只是加锁的部分（修改共享数据的部分）是串行的，两者从保护数据安全方面来说是一样的，但是加锁的效率更高。
from threading import current_thread,Thread,Lock import os,time def task(): time.sleep(3) print('%s start to run' %current_thread().getName()) global n temp=n time.sleep(0.5) n=temp-1 if __name__ == '__main__': n=100 lock=Lock() start_time=time.time() for i in range(100): t=Thread(target=task) t.start() t.join() stop_time=time.time() print('主:%s n:%s' %(stop_time-start_time,n)) ''' Thread-1 start to run Thread-2 start to run ...... Thread-100 start to run 主:350.6937336921692 n:0 #耗时是多么的恐怖 '''
九、死锁现象与递归锁
死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程。
from threading import Thread, Lock import time # 实例化两把锁 mutexA = Lock() mutexB = Lock() class MyThread(Thread): def run(self): self.f1() self.f2() def f1(self): mutexA.acquire() print("%s 拿到A锁" % self.name) mutexB.acquire() print("%s 拿到了B锁" % self.name) mutexB.release() mutexA.release() def f2(self): mutexB.acquire() print("%s 拿到B锁" % self.name) time.sleep(0.1) # 线程1在此休息0.1秒 mutexA.acquire() print("%s 拿到了A锁" % self.name) mutexA.release() mutexB.release() if __name__ == '__main__': for i in range(10): t = MyThread() t.start() # 信号提交，就几乎立马启动了 """程序输出下面内容后，卡住了 Thread-1 拿到A锁 Thread-1 拿到了B锁 Thread-1 拿到B锁 Thread-2 拿到A锁 ————》线程1睡着时，线程2拿到A锁，要去拿B锁，B锁在线程1手里，线程1睡完要去拿A锁，A锁在线程2手里，因此产生死锁。 """
线程1睡着时，线程2拿到A锁，要去拿B锁，B锁在线程1手里，线程1睡完要去拿A锁，A锁在线程2手里，因此产生死锁。上述例子也说明：自己处理锁其实非常繁琐也非常危险，一定要在适当的时候考虑把锁释放掉。处理不当就会出现死锁，整个程序就会卡在原地。
这是由于互斥锁只能acquire一次，使用方法如下：
from threading import Thread, Lock mutexA = Lock() mutexA.acquire() mutexA.release()
解决办法——递归锁，可以连续acquire多次，每acquire一次计数器加1；只要计数不为0，就不能被其他线程抢到（只有计数为0，才能被抢到acquire）
在Python中为了支持同一线程中多次请求同一资源，提供了可重入锁RLock。
这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：
mutexA=mutexB=threading.RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1
，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止
from threading import Thread, RLock import time """链式赋值""" mutexA=mutexB=RLock() # 使用递归锁可以 class MyThread(Thread): def run(self): self.f1() self.f2() def f1(self): mutexA.acquire() # 递归锁计数器加1 print("%s 拿到A锁" % self.name) mutexB.acquire() print("%s 拿到了B锁" % self.name) mutexB.release() # 递归锁计数器减1 mutexA.release() def f2(self): mutexB.acquire() print("%s 拿到B锁" % self.name) time.sleep(1) # 线程1在此休息0.1秒 mutexA.acquire() print("%s 拿到了A锁" % self.name) mutexA.release() mutexB.release() # 计数为0，其他线程可以抢acquire if __name__ == '__main__': for i in range(10): t = MyThread() t.start() # 信号提交，就几乎立马启动了 """ 第一个线程计数器为0 后，其他线程可以开始抢acquire，因此顺序是不固定的。 Thread-1 拿到A锁 Thread-1 拿到了B锁 Thread-1 拿到B锁 Thread-1 拿到了A锁 Thread-2 拿到A锁 Thread-2 拿到了B锁 Thread-2 拿到B锁 Thread-2 拿到了A锁 Thread-4 拿到A锁 Thread-4 拿到了B锁 Thread-4 拿到B锁 Thread-4 拿到了A锁 Thread-6 拿到A锁 Thread-6 拿到了B锁 Thread-6 拿到B锁 Thread-6 拿到了A锁 Thread-8 拿到A锁 Thread-8 拿到了B锁 Thread-8 拿到B锁 Thread-8 拿到了A锁 Thread-10 拿到A锁 Thread-10 拿到了B锁 Thread-10 拿到B锁 Thread-10 拿到了A锁 Thread-5 拿到A锁 Thread-5 拿到了B锁 Thread-5 拿到B锁 Thread-5 拿到了A锁 Thread-9 拿到A锁 Thread-9 拿到了B锁 Thread-9 拿到B锁 Thread-9 拿到了A锁 Thread-7 拿到A锁 Thread-7 拿到了B锁 Thread-7 拿到B锁 Thread-7 拿到了A锁 Thread-3 拿到A锁 Thread-3 拿到了B锁 Thread-3 拿到B锁 Thread-3 拿到了A锁 """
十、信号量semaphore
信号量也是一把锁，可以指定信号量为5，对比互斥锁同一时间只能有一个任务抢到锁去执行，信号量同一时间可以有5个任务拿到锁去执行，如果说互斥锁是合租房屋的人去抢一个厕所，那么信号量就相当于一群路人争抢公共厕所，公共厕所有多个坑位，这意味着同一时间可以有多个人上公共厕所，但公共厕所容纳的人数是一定的，这便是信号量的大小。
from threading import Thread, Semaphore, currentThread import time, random sm = Semaphore(3) # 定义出坑的个数 def task(): # sm.acquire() # print("%s in" % currentThread().getName()) # sm.release() with sm: print("%s in " % currentThread().getName()) time.sleep(random.randint(1, 3)) if __name__ == '__main__': for i in range(10): t = Thread(target=task) t.start() """ Thread-1 in Thread-2 in Thread-3 in Thread-5 in Thread-6 in Thread-4 in Thread-7 in Thread-8 in Thread-9 in Thread-10 in """
Semaphore管理一个内置的计算器，每当调用acquire()时内置计数器-1；调用release()时内置计数器+1；计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。
与进程池是完全不同的概念，进程池Pool(4)，最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一堆线程/进程
互斥锁与信号量推荐博客：http://url.cn/5DMsS9r
十一、Event
同进程的一样线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。
为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行。
from threading import Event ：调用event event.isSet()：返回event的状态值； event.wait()：如果 event.isSet()==False将阻塞线程； event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度； event.clear()：恢复event的状态值为False。 ————set()后再clear()，重置到初始状态。
例如，有多个工作线程尝试链接MySQL，我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作。
from threading import Thread, Event import time event = Event() def student(name): print("学生%s正在听课" % name) event.wait() # 在原地等待 print("学生%s课间活动" % name) def teacher(name): print("老师%s 正在授课" % name) time.sleep(7) event.set() # 这个运行后等待结束 if __name__ == '__main__': stu1 = Thread(target=student, args=('alex',)) stu2 = Thread(target=student, args=('wxx',)) stu3 = Thread(target=student, args=('yxx',)) t1 = Thread(target=teacher, args=('egon',)) stu1.start() stu2.start() stu3.start() t1.start() """ 学生alex正在听课 学生wxx正在听课 学生yxx正在听课 老师egon 正在授课 ------->在这里等7秒后，学生开始做课间活动 学生alex课间活动 学生wxx课间活动 学生yxx课间活动 """
将上例改写：有的学生线程，需要在老师发出结束信号前就去做其他工作。
from threading import Thread, Event import time event = Event() def student(name): print("学生%s正在听课" % name) event.wait(2) print("学生%s课间活动" % name) def teacher(name): print("老师%s 正在授课" % name) time.sleep(7) event.set() if __name__ == '__main__': stu1 = Thread(target=student, args=('alex',)) stu2 = Thread(target=student, args=('wxx',)) stu3 = Thread(target=student, args=('yxx',)) t1 = Thread(target=teacher, args=('egon',)) stu1.start() stu2.start() stu3.start() t1.start() """ 学生alex正在听课 学生wxx正在听课 学生yxx正在听课 老师egon 正在授课 -------》等两秒后，学生就去做课间活动了，等满七秒，程序才结束 学生alex课间活动 学生yxx课间活动 学生wxx课间活动 """
有很多时候，多次检测不成功，需要设置超时时间：
from threading import Thread, Event, currentThread import time event = Event() def conn(): n=0 while not event.is_set(): # 还没有set()过，值为False if n == 3: print("%s try too many times!" % currentThread().getName()) return # 结束整个函数，如果break，则链接成功了 print("%s try %s" % (currentThread().getName(), n)) event.wait(0.5) # 原地等待，并设置了超时时间 n += 1 print("%s is connected" % currentThread().getName()) def check(): print("%s is checking" % currentThread().getName()) time.sleep(5) # 模拟检测 event.set() # 检测OK if __name__ == '__main__': for i in range(3): t = Thread(target=conn) t.start() t = Thread(target=check) # 检测线程 t.start() """ Thread-1 try 0 Thread-2 try 0 Thread-3 try 0 Thread-4 is checking Thread-1 try 1 Thread-2 try 1 Thread-3 try 1 Thread-1 try 2 Thread-3 try 2 Thread-2 try 2 Thread-1 try too many times! Thread-2 try too many times! Thread-3 try too many times! """
十二、条件Condition
使得线程等待，只有满足某条件时，才释放n个线程
import threading def run(n): con.acquire() con.wait() print("run the thread: %s" %n) con.release() if __name__ == '__main__': con = threading.Condition() for i in range(10): t = threading.Thread(target=run, args=(i,)) t.start() while True: inp = input('>>>') if inp == 'q': break con.acquire() con.notify(int(inp)) con.release()
函数化：
def condition_func(): ret = False inp = input('>>>') if inp == '1': ret = True return ret def run(n): con.acquire() con.wait_for(condition_func) print("run the thread: %s" %n) con.release() if __name__ == '__main__': con = threading.Condition() for i in range(10): t = threading.Thread(target=run, args=(i,)) t.start()
十三、定时器
定时器：指定时间后，隔多少时间之后触发去执行某操作。
from threading import Timer def task(name): print("hello %s" % name) t = Timer(5, task, args=('egon',)) # 创建对象，Timer是Thread的子类，其实就是一个线程 t.start() # hello egon ---->在等五秒后打印
import random def make_code(n=4): # 设置默认值为4 res = '' for i in range(n): s1 = str(random.randint(0,9)) # 随机数字字符 s2 = chr(random.randint(65, 90)) # 随机字母 ，注意了解chr()内置函数 res += random.choice([s1, s2]) return res print(make_code()) """ 6HS8 6S38 ————》结果随机产生 """
将定时器改写为类：
from threading import Timer import random class Code: def __init__(self): self.make_cache() def make_cache(self, interval=8): self.cache = self.make_code() # 缓存验证码 print(self.cache) self.t = Timer(interval, self.make_cache) # 创建定时器，到时间刷新一次 self.t.start() def make_code(self, n=4): # 设置默认值为4 res = '' for i in range(n): s1 = str(random.randint(0,9)) # 随机数字字符 s2 = chr(random.randint(65, 90)) # 随机字母 ，注意了解chr()内置函数 res += random.choice([s1, s2]) return res def check(self): while True: code = input("请输入你的验证码>>：").strip() if code.upper() == self.cache: print("验证码输入正确") self.t.cancel() break obj = Code() obj.check() """ E8I2 请输入你的验证码>>：E8I2 验证码输入正确 """
十四、线程queue
queue队列 ：使用import queue，用法与进程Queue一样
q.put方法用以插入数据到队列中。
q.get方法可以从队列读取并且删除一个元素。
1、class queue.Queue(maxsize) 定义好队列存数据的最大值，队列存取规则是先进先出
import queue q = queue.Queue(3) # 生成队列，存数据最大值是3 先进先出 q.put("first") # 放值进去 q.put(2) q.put("third") # q.put(4) # 队列满了，阻塞 print(q.get()) # 取数据 print(q.get()) print(q.get()) """ first 2 third """
队列放满了继续放，会造成阻塞；
如果把默认的block=True改为False，程序会直接报错：raise full；
如果不修改block，而是设置timeout，程序会等到timeout时间之后报错raise full。
队列为空继续取，和上述的情况非常类似：
import queue q = queue.Queue(3) # 生成队列，存数据最大值是3 先进先出 q.put("first") # 放值进去 q.put(2) q.put("third") # q.put(4) # 队列满了，阻塞 # q.put(4, block=False) # 默认是block=True, 改为False后，队列满了还加数据，程序报错raise Full queue.Full # q.put(4, block=True, timeout=3) # 设置了block=True队列满不会直接报错了，但是还加上了timeout=3,程序会等3秒后提示报错queue.Full # 同理get()方法也有这些参数 print(q.get()) # 取数据 print(q.get()) print(q.get()) # print(q.get(block=False)) # 在队列空，还取时，一般是卡住，但加入了block=False参数的话，会提示报错queue.Empty print(q.get_nowait()) # 这个效果同上 print(q.get(block=True, timeout=3)) # 队列空，还取数据时，会按照timeout时间等待，到时间后提示queue.Empty
2、class queue.LifeQueue(maxsize) 定义为堆栈，堆栈的存取规则是后进先出
import queue q = queue.LifoQueue(3) # 堆栈 q.put("first") q.put(2) q.put("third") print(q.get()) # third print(q.get()) # 2 print(q.get()) # first
3、class queue.PriorityQueue(maxsize) 存储数据时可设置优先级的队列
import queue q = queue.PriorityQueue(3) q.put(10, 'one') q.put(40, 'two') q.put(30, 'three') print(q.get()) # 10 print(q.get()) # 30 print(q.get()) # 40
数字越小优先级越高（取出的优先级）。
十五、进程池和线程池
在刚开始学多进程或多线程时，我们迫不及待地基于多进程或多线程实现并发的套接字通信：
# -*- coding:utf-8 -*- __author__ = 'Qiushi Huang' from socket import * from threading import Thread # 通讯和建立链接分开，启动不同的线程，大家是并发执行。 def communicate(conn): while True: try: data = conn.recv(1024) if not data:break conn.send(data.upper()) except ConnectionResetError: break conn.close() def server(ip, port): server = socket(AF_INET, SOCK_STREAM) server.bind((ip, port)) server.listen(5) while True: conn, addr = server.accept() # 建链接 t = Thread(target=communicate, args=(conn,)) # 建一个链接创一个线程 t.start() # communicate(conn) server.close() if __name__ == '__main__': server('127.0.0.1', 8091) # 主线程 """ 这种解决方案的问题是：当客户端越来越多后，线程也会越来越多，会带来服务崩溃的问题。 """
# -*- coding:utf-8 -*- __author__ = 'Qiushi Huang' # 使用时，可以一个程序运行多次，这是多个不同的in from socket import * client = socket(AF_INET, SOCK_STREAM) client.connect(("127.0.0.1", 8091)) while True: msg = input(">>").strip() if not msg:continue client.send(msg.encode("utf-8")) data = client.recv(1024) print(data.decode("utf-8")) client.close()
这种实现方式的致命缺陷是：服务的开启的进程数或线程数都会随着并发的客户端数目地增多而增多，这会对服务端主机带来巨大的压力，甚至于不堪重负而瘫痪，于是我们必须对服务端开启的进程数或线程数加以控制，让机器在一个自己可以承受的范围内运行，这就是进程池或线程池的用途，例如进程池，就是用来存放进程的池子，本质还是基于多进程，只不过是对开启进程的数目加上了限制。
进程池和线程池的接口一模一样，用法也完全相同。池就是要对数目加以限制，保证机器一个可承受的范围，以一个健康的状态保证它的运行。
1、介绍：
官网：https://docs.python.org/dev/library/concurrent.futures.html concurrent.futures 模块提供了高度封装的异步调用接口 ThreadPoolExecutor： 线程池，提供异步调用 ProcessPoolExecutor: 进程池，提供异步调用 Both implement the same interface, which is defined by the abstract Executor class.
基本方法：
1、submit(fn, *args, **kwargs) 异步提交任务 2、map(func, *iterables, timeout=None, chunksize=1) 取代for循环submit的操作 3、shutdown(wait=True) 相当于进程池的pool.close()+pool.join()操作 wait=True，等待池内所有任务执行完毕回收完资源后才继续 wait=False，立即返回，并不会等待池内的任务执行完毕 但不管wait参数为何值，整个程序都会等到所有任务执行完毕 submit和map必须在shutdown之前 4、result(timeout=None) 取得结果 5、add_done_callback(fn) 回调函数
2、进程池：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor import os, time, random def task(name): print("name: %s pid: %s run" % (name, os.getpid())) time.sleep(random.randint(1,3)) if __name__ == '__main__': pool = ProcessPoolExecutor(4) # 指定进程池大小，最大进程数，如果不指定默认是CPU核数 for i in range(10): """从始至终四个进程解决这10个任务，谁没事了接新任务""" pool.submit(task, 'egon%s' %i) # 提交任务的方式————异步调用：提交完任务，不用在原地等任务执行拿到结果。 print("主进程") """ name: egon0 pid: 12445 run name: egon1 pid: 12444 run name: egon2 pid: 12446 run name: egon3 pid: 12447 run 主进程 name: egon4 pid: 12445 run name: egon5 pid: 12444 run name: egon6 pid: 12446 run name: egon7 pid: 12445 run name: egon8 pid: 12446 run name: egon9 pid: 12447 run """
shutdown()方法的使用：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor import os, time, random def task(name): print("name: %s pid: %s run" % (name, os.getpid())) time.sleep(random.randint(1,3)) if __name__ == '__main__': pool = ProcessPoolExecutor(4) # 指定进程池大小，最大进程数，如果不指定默认是CPU核数 for i in range(10): """从始至终四个进程解决这10个任务，谁没事了接新任务""" pool.submit(task, 'egon%s' %i) # 提交任务的方式————异步调用：提交完任务，不用在原地等任务执行拿到结果。 pool.shutdown() # 把提交任务入口关闭，默认参数wait=True；同时还进行了pool.join()操作，等任务提交结束，再结束主进程 print("主进程") """ name: egon0 pid: 12502 run name: egon1 pid: 12503 run name: egon2 pid: 12504 run name: egon3 pid: 12505 run name: egon4 pid: 12502 run name: egon5 pid: 12503 run name: egon6 pid: 12505 run name: egon7 pid: 12504 run name: egon8 pid: 12503 run name: egon9 pid: 12505 run 主进程 """
3、针对线程的情况：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor import os, time, random def task(name): print("name: %s pid: %s run" % (name, os.getpid())) time.sleep(random.randint(1,3)) if __name__ == '__main__': pool = ThreadPoolExecutor(4) for i in range(10): """从始至终四个进程解决这10个任务，谁没事了接新任务""" pool.submit(task, 'egon%s' %i) # 提交任务的方式————异步调用：提交完任务，不用在原地等任务执行拿到结果。 pool.shutdown(wait=True) # 把提交任务入口关闭，默认参数wait=True；同时还进行了pool.join()操作，等任务提交结束，再结束主进程 print("主进程") """ name: egon0 pid: 12528 run name: egon1 pid: 12528 run name: egon2 pid: 12528 run name: egon3 pid: 12528 run name: egon4 pid: 12528 run name: egon5 pid: 12528 run name: egon6 pid: 12528 run name: egon7 pid: 12528 run name: egon8 pid: 12528 run name: egon9 pid: 12528 run 主进程 """
currentThread()方法查看线程名：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor from threading import currentThread # 查看线程名 import os, time, random def task(): print("name: %s pid: %s run" % (currentThread().getName(), os.getpid())) time.sleep(random.randint(1,3)) if __name__ == '__main__': pool = ThreadPoolExecutor(4) for i in range(10): """从始至终四个进程解决这10个任务，谁没事了接新任务""" pool.submit(task,) # 提交任务的方式————异步调用：提交完任务，不用在原地等任务执行拿到结果。 pool.shutdown(wait=True) # 把提交任务入口关闭，默认参数wait=True；同时还进行了pool.join()操作，等任务提交结束，再结束主进程 print("主进程") """ name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_0 pid: 12554 run name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_1 pid: 12554 run name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_2 pid: 12554 run name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_3 pid: 12554 run name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_2 pid: 12554 run name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_0 pid: 12554 run name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_3 pid: 12554 run name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_2 pid: 12554 run name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_1 pid: 12554 run name: <concurrent.futures.thread.ThreadPoolExecutor object at 0x10401af28>_0 pid: 12554 run 主进程 """
4、map方法
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor import os,time,random def task(n): print('%s is runing' %os.getpid()) time.sleep(random.randint(1,3)) return n**2 if __name__ == '__main__': executor=ThreadPoolExecutor(max_workers=3) # for i in range(11): # future=executor.submit(task,i) executor.map(task,range(1,12)) #map取代了for+submit
5、异步调用和回调机制
同步调用：提交完任务后，就在原地等待任务执行完毕，拿到执行结果，再执行下一行。
导致程序是串行执行。
from concurrent.futures import ThreadPoolExecutor import time import random def la(name): print("%s is laing" % name) time.sleep(random.randint(3,5)) res = random.randint(7, 13)*'#' return {'name':name, 'res':res} def weigh(shit): name = shit['name'] size = len(shit['res']) print('%s 拉了 《%s》kg' % (name, size)) if __name__ == '__main__': pool = ThreadPoolExecutor(13) shit1 = pool.submit(la, 'alex').result() weigh(shit1) shit2 = pool.submit(la, 'wupeiqi').result() weigh(shit2) shit3 = pool.submit(la, "yuanhao").result() weigh(shit3) """ alex is laing alex 拉了 《7》kg wupeiqi is laing wupeiqi 拉了 《9》kg yuanhao is laing yuanhao 拉了 《11》kg """
异步调用：与同步相对，一个异步功能调用提交完任务后，调用者不会立刻得到结果，而在完成后，通过状态、通知或回调函数来通知调用者。
from concurrent.futures import ThreadPoolExecutor import time import random def la(name): print("%s is laing" % name) time.sleep(random.randint(3,5)) res = random.randint(7, 13)*'#' # return {'name':name, 'res':res} weigh({'name':name, 'res':res}) # 直接把字典传给称重weigh()，但造成了程序耦合 def weigh(shit): name = shit['name'] size = len(shit['res']) print('%s 拉了 《%s》kg' % (name, size)) if __name__ == '__main__': pool = ThreadPoolExecutor(13) pool.submit(la, 'alex') pool.submit(la, 'wupeiqi') pool.submit(la, "yuanhao") """并发执行拉的任务，谁执行完，谁把结果传给称重功能。 alex is laing wupeiqi is laing yuanhao is laing alex 拉了 《10》kg wupeiqi 拉了 《7》kg yuanhao 拉了 《7》kg """
可以为进程池或线程池内的每个进程或线程绑定一个函数，该函数在进程或线程的任务执行完毕后自动触发，并接收任务的返回值当作参数，该函数称为回调函数。
from concurrent.futures import ThreadPoolExecutor import time import random def la(name): print("%s is laing" % name) time.sleep(random.randint(3,5)) res = random.randint(7, 13)*'#' return {'name':name, 'res':res} # weigh({'name':name, 'res':res}) # 直接把字典传给称重weigh()，造成了程序耦合 def weigh(shit): shit = shit.result() # 对象.result()拿到结果并赋值给shit name = shit['name'] size = len(shit['res']) print('%s 拉了 《%s》kg' % (name, size)) if __name__ == '__main__': pool = ThreadPoolExecutor(13) # 回调函数，前面任务执行完，return返回值就会自动触发weith功能执行，把pool.submit(la, 'alex')对象当做参数传给weigh() pool.submit(la, 'alex').add_done_callback(weigh) pool.submit(la, 'wupeiqi').add_done_callback(weigh) pool.submit(la, "yuanhao").add_done_callback(weigh) """ alex is laing wupeiqi is laing yuanhao is laing alex 拉了 《10》kg wupeiqi 拉了 《7》kg yuanhao 拉了 《7》kg """
阻塞和非阻塞的区别？
阻塞是进程运行的一种状态，遇到I/O就会进入阻塞状态，会被剥夺走CPU的执行权限。
阻塞调用在调用结果返回前，当前线程会被挂起。直到得到返回结果，才会将阻塞线程激活。
非阻塞调用：指在不能立刻得到结果之前也会立刻返回，同时函数不会阻塞当前线程。
阻塞和同步调用的区别？
同步调用：调用会一直等待，直到任务返回结果为止，即使被抢走cpu也是处于就绪状态。
阻塞调用：socket工作在阻塞模式时，如果没有数据下调用recv()，当前进程挂起，直到有数据为止。
同步异步针对的是函数/任务的调用方式；阻塞非阻塞针对的是进程或线程。
# -*- coding:utf-8 -*- __author__ = 'Qiushi Huang' # 多I/O的问题采用线程池 from concurrent.futures import ThreadPoolExecutor import requests import time def get(url): print('get %s' % url) # requests.get()就是目标页面下载一个文件到本地来 response = requests.get(url) # 对象 time.sleep(3) # 模拟网络延迟 # print(response.text) # 网页内容 return {'url':url, 'content':response.text} def paese(res): # 解析，正则表达式 res = res.result() print('%s parse res is %s' % (res['url'], len(res['content']))) if __name__ == '__main__': urls = [ 'http://www.cnblogs.com/linhaifeng', 'http://www.python.org', 'http://www.openstack.org' ] pool = ThreadPoolExecutor(2) for url in urls: pool.submit(get, url).add_done_callback(paese) # 回调函数 """ get http://www.cnblogs.com/linhaifeng get http://www.python.org ——————》明显的等的效果 http://www.cnblogs.com/linhaifeng parse res is 16320 get http://www.openstack.org http://www.python.org parse res is 49014 http://www.openstack.org parse res is 63429 """
多IO问题采用线程池。
由上例可以分析出异步调用加回调机制使用的场景。
6、线程池优化实现并发的套接字
# -*- coding:utf-8 -*- __author__ = 'Qiushi Huang' """ 原先多线程解决方案的问题是：当客户端越来越多后，线程也会越来越多，会带来服务崩溃的问题。 不应该随着客户端数量增加不断地增加线程，需要基于线程池实现，限制线程数量 """ from socket import * from concurrent.futures import ThreadPoolExecutor def communicate(conn): while True: try: data = conn.recv(1024) # if not data:continue # 这里卡了很久，需要注意,这种情况下关闭客户端，线程池没有减少 # if data.decode('utf-8') == 'q':break # 测试，这种情况下，线程池减少，新进程加入进程池 if not data:break conn.send(data.upper()) except ConnectionResetError: break conn.close() def server(ip, port): server = socket(AF_INET, SOCK_STREAM) server.bind((ip, port)) server.listen(5) while True: conn, addr = server.accept() # 建链接 pool.submit(communicate, conn) server.close() if __name__ == '__main__': pool = ThreadPoolExecutor(2) # 一般写机器可承受的范围内 server('127.0.0.1', 8092) # 主线程
# -*- coding:utf-8 -*- __author__ = 'Qiushi Huang' from socket import * client = socket(AF_INET, SOCK_STREAM) client.connect(("127.0.0.1", 8092)) while True: msg = input(">>").strip() if not msg:continue client.send(msg.encode("utf-8")) data = client.recv(1024) print(data.decode("utf-8")) client.close()