python2个子线程等待_python并发编程之多线程2死锁与递归锁，信号量等...

一、死锁现象与递归锁

进程也是有死锁的

所谓死锁&＃xff1a; 是指两个或两个以上的进程或线程在执行过程中&＃xff0c;因争夺资源而造成的一种互相等待的现象&＃xff0c;若无外力作用&＃xff0c;

它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁&＃xff0c;这些永远在互相等待的进程称为死锁进程&＃xff0c;

如下就是死锁

1 死锁-------------------

2 from threading importThread,Lock,RLock3 importtime4 mutexA &＃61;Lock()5 mutexB &＃61;Lock()6 classMyThread(Thread):7 defrun(self):8 self.f1()9 self.f2()10 deff1(self):11 mutexA.acquire()12 print(&＃39;\033[33m%s 拿到A锁&＃39;%self.name)13 mutexB.acquire()14 print(&＃39;\033[45%s 拿到B锁&＃39;%self.name)15 mutexB.release()16 mutexA.release()17 deff2(self):18 mutexB.acquire()19 print(&＃39;\033[33%s 拿到B锁&＃39; %self.name)20 time.sleep(1) #睡一秒就是为了保证A锁已经被别人那到了

21 mutexA.acquire()22 print(&＃39;\033[45m%s 拿到B锁&＃39; %self.name)23 mutexA.release()24 mutexB.release()25 if __name__ &＃61;&＃61; &＃39;__main__&＃39;:26 for i in range(10):27 t &＃61;MyThread()28 t.start() #一开启就会去调用run方法

死锁现象

那么怎么解决死锁现象呢&＃xff1f;

解决方法&＃xff0c;递归锁&＃xff1a;在Python中为了支持在同一线程中多次请求同一资源&＃xff0c;python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量&＃xff0c;counter记录了acquire的次数&＃xff0c;从而使得资源可以被多次require。

直到一个线程所有的acquire都被release&＃xff0c;其他的线程才能获得资源。上面的例子如果使用RLock代替Lock&＃xff0c;则不会发生死锁

mutexA&＃61;mutexB&＃61;threading.RLock() #一个线程拿到锁&＃xff0c;counter加1,该线程内又碰到加锁的情况&＃xff0c;

则counter继续加1&＃xff0c;这期间所有其他线程都只能等待&＃xff0c;等待该线程释放所有锁&＃xff0c;即counter递减到0为止

1 #2.解决死锁的方法--------------递归锁

2 from threading importThread,Lock,RLock3 importtime4 mutexB &＃61; mutexA &＃61;RLock()5 classMyThread(Thread):6 defrun(self):7 self.f1()8 self.f2()9 deff1(self):10 mutexA.acquire()11 print(&＃39;\033[33m%s 拿到A锁&＃39;%self.name)12 mutexB.acquire()13 print(&＃39;\033[45%s 拿到B锁&＃39;%self.name)14 mutexB.release()15 mutexA.release()16 deff2(self):17 mutexB.acquire()18 print(&＃39;\033[33%s 拿到B锁&＃39; %self.name)19 time.sleep(1) #睡一秒就是为了保证A锁已经被别人拿到了

20 mutexA.acquire()21 print(&＃39;\033[45m%s 拿到B锁&＃39; %self.name)22 mutexA.release()23 mutexB.release()24 if __name__ &＃61;&＃61; &＃39;__main__&＃39;:25 for i in range(10):26 t &＃61;MyThread()27 t.start() #一开启就会去调用run方法

解决死锁

二、信号量Semaphore(其实也是一把锁)

Semaphore管理一个内置的计数器

Semaphore与进程池看起来类似&＃xff0c;但是是完全不同的概念。

进程池&＃xff1a;Pool(4)&＃xff0c;最大只能产生四个进程&＃xff0c;而且从头到尾都只是这四个进程&＃xff0c;不会产生新的。

信号量&＃xff1a;信号量是产生的一堆进程/线程&＃xff0c;即产生了多个任务都去抢那一把锁

1 from threading importThread,Semaphore,currentThread2 importtime,random3 sm &＃61; Semaphore(5) #运行的时候有5个人

4 deftask():5 sm.acquire()6 print(&＃39;\033[42m %s上厕所&＃39;%currentThread().getName())7 time.sleep(random.randint(1,3))8 print(&＃39;\033[31m %s上完厕所走了&＃39;%currentThread().getName())9 sm.release()10 if __name__ &＃61;&＃61; &＃39;__main__&＃39;:11 for i in range(20): #开了10个线程 &＃xff0c;这20人都要上厕所

12 t &＃61; Thread(target&＃61;task)13 t.start()

Semaphore举例

1 hread-1上厕所2 Thread-2上厕所3 Thread-3上厕所4 Thread-4上厕所5 Thread-5上厕所6 Thread-3上完厕所走了7 Thread-6上厕所8 Thread-1上完厕所走了9 Thread-7上厕所10 Thread-2上完厕所走了11 Thread-8上厕所12 Thread-6上完厕所走了13 Thread-5上完厕所走了14 Thread-4上完厕所走了15 Thread-9上厕所16 Thread-10上厕所17 Thread-11上厕所18 Thread-9上完厕所走了19 Thread-12上厕所20 Thread-7上完厕所走了21 Thread-13上厕所22 Thread-10上完厕所走了23 Thread-8上完厕所走了24 Thread-14上厕所25 Thread-15上厕所26 Thread-12上完厕所走了27 Thread-11上完厕所走了28 Thread-16上厕所29 Thread-17上厕所30 Thread-14上完厕所走了31 Thread-15上完厕所走了32 Thread-17上完厕所走了33 Thread-18上厕所34 Thread-19上厕所35 Thread-20上厕所36 Thread-13上完厕所走了37 Thread-20上完厕所走了38 Thread-16上完厕所走了39 Thread-18上完厕所走了40 Thread-19上完厕所走了

运行结果

三、Event

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行

from threading import Event

Event.isSet() #返回event的状态值

Event.wait() #如果 event.isSet()&＃61;&＃61;False将阻塞线程&＃xff1b;

Event.set() #设置event的状态值为True&＃xff0c;所有阻塞池的线程激活进入就绪状态&＃xff0c; 等待操作系统调度&＃xff1b;

Event.clear() #恢复

例如1.&＃xff0c;有多个工作线程尝试链接MySQL&＃xff0c;我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器&＃xff0c;如果连接不成功&＃xff0c;都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作

1 #首先定义两个函数&＃xff0c;一个是连接数据库

2 #一个是检测数据库

3 from threading importThread,Event,currentThread4 importtime5 e &＃61;Event()6 defconn_mysql():7 &＃39;&＃39;&＃39;链接数据库&＃39;&＃39;&＃39;

8 count &＃61; 1

9 while not e.is_set(): #当没有检测到时候

10 if count >3: #如果尝试次数大于3&＃xff0c;就主动抛异常

11 raise ConnectionError(&＃39;尝试链接的次数过多&＃39;)12 print(&＃39;\033[45m%s 第%s次尝试&＃39;%(currentThread(),count))13 e.wait(timeout&＃61;1) #等待检测(里面的参数是超时1秒)

14 count&＃43;&＃61;1

15 print(&＃39;\033[44m%s 开始链接...&＃39;%(currentThread().getName()))16 defcheck_mysql():17 &＃39;&＃39;&＃39;检测数据库&＃39;&＃39;&＃39;

18 print(&＃39;\033[42m%s 检测mysql...&＃39; %(currentThread().getName()))19 time.sleep(5)20 e.set()21 if __name__ &＃61;&＃61; &＃39;__main__&＃39;:22 for i in range(3): #三个去链接

23 t &＃61; Thread(target&＃61;conn_mysql)24 t.start()25 t &＃61; Thread(target&＃61;check_mysql)26 t.start()

详看

2.例如2&＃xff0c;红绿灯的例子

1 from threading importThread,Event,currentThread2 importtime3 e &＃61;Event()4 deftraffic_lights():5 &＃39;&＃39;&＃39;红绿灯&＃39;&＃39;&＃39;

6 time.sleep(5)7 e.set()8 defcar():9 &＃39;&＃39;&＃39;车&＃39;&＃39;&＃39;

10 print(&＃39;\033[42m %s 等绿灯\033[0m&＃39;%currentThread().getName())11 e.wait()12 print(&＃39;\033[44m %s 车开始通行&＃39; %currentThread().getName())13 if __name__ &＃61;&＃61; &＃39;__main__&＃39;:14 for i in range(10):15 t &＃61; Thread(target&＃61;car) #10辆车

16 t.start()17 traffic_thread &＃61; Thread(target&＃61;traffic_lights) #一个红绿灯

18 traffic_thread.start()

红绿灯

四、定时器(Timer)

指定n秒后执行某操作

from threading importTimerdeffunc(n):print(&＃39;hello,world&＃39;,n)

t&＃61; Timer(3,func,args&＃61;(123,)) #等待三秒后执行func函数&＃xff0c;因为func函数有参数&＃xff0c;那就再传一个参数进去

t.start()

五、线程queue

queue队列 &＃xff1a;使用import queue&＃xff0c;用法与进程Queue一样

queue.Queue(maxsize&＃61;0) #先进先出

1 #1.队列-----------

2 importqueue3 q &＃61; queue.Queue(3) #先进先出

4 q.put(&＃39;first&＃39;)5 q.put(&＃39;second&＃39;)6 q.put(&＃39;third&＃39;)7 print(q.get())8 print(q.get())9 print(q.get())

View Code

queue.LifoQueue(maxsize&＃61;0)#先进后出

1 #2.堆栈----------

2 q &＃61; queue.LifoQueue() #先进后出(或者后进先出)

3 q.put(&＃39;first&＃39;)4 q.put(&＃39;second&＃39;)5 q.put(&＃39;third&＃39;)6 q.put(&＃39;for&＃39;)7 print(q.get())8 print(q.get())9 print(q.get())

View Code

queue.PriorityQueue(maxsize&＃61;0) #存储数据时可设置优先级的队列

1 #----------------

2 &＃39;&＃39;&＃39;3.put进入一个元组&＃xff0c;元组的第一个元素是优先级3 (通常也可以是数字&＃xff0c;或者也可以是非数字之间的比较)4 数字越小&＃xff0c;优先级越高&＃39;&＃39;&＃39;

5 q &＃61;queue.PriorityQueue()6 q.put((20,&＃39;a&＃39;))7 q.put((10,&＃39;b&＃39;)) #先出来的是b,数字越小优先级越高嘛

8 q.put((30,&＃39;c&＃39;))9 print(q.get())10 print(q.get())11 print(q.get())

View Code

六、多线程性能测试

1.多核也就是多个CPU

(1)cpu越多&＃xff0c;提高的是计算的性能

(2)如果程序是IO操作的时候(多核和单核是一样的)&＃xff0c;再多的cpu也没有意义。

2.实现并发

第一种&＃xff1a;一个进程下&＃xff0c;开多个线程

第二种&＃xff1a;开多个进程

3.多进程&＃xff1a;

优点&＃xff1a;可以利用多核

缺点&＃xff1a;开销大

4.多线程

优点&＃xff1a;开销小

缺点&＃xff1a;不可以利用多核

5多进程和多进程的应用场景

1.计算密集型&＃xff1a;也就是计算多&＃xff0c;IO少

如果是计算密集型&＃xff0c;就用多进程(如金融分析等)

2.IO密集型&＃xff1a;也就是IO多&＃xff0c;计算少

如果是IO密集型的&＃xff0c;就用多线程(一般遇到的都是IO密集型的)

下例子练习&＃xff1a;

1 #计算密集型的要开启多进程

2 from multiprocessing importProcess3 from threading importThread4 importtime5 defwork():6 res &＃61;07 for i in range(10000000):8 res&＃43;&＃61;i9 if __name__ &＃61;&＃61; &＃39;__main__&＃39;:10 l &＃61;[]11 start &＃61;time.time()12 for i in range(4):13 p &＃61; Process(target&＃61;work) #1.9371106624603271 #可以利用多核(也就是多个cpu)

14 #p &＃61; Thread(target&＃61;work) #3.0401737689971924

15 l.append(p)16 p.start()17 for p inl:18 p.join()19 stop &＃61;time.time()20 print(&＃39;%s&＃39;%(stop-start))

计算密集型

1 #I/O密集型要开启多线程

2 from multiprocessing importProcess3 from threading importThread4 importtime5 defwork():6 time.sleep(3)7 if __name__ &＃61;&＃61; &＃39;__main__&＃39;:8 l &＃61;[]9 start &＃61;time.time()10 for i in range(400):11 #p &＃61; Process(target&＃61;work) #34.9549994468689 #因为开了好多进程&＃xff0c;它的开销大&＃xff0c;花费的时间也就长了

12 p &＃61; Thread(target&＃61;work) #2.2151265144348145 #当开了多个线程的时候&＃xff0c;它的开销小&＃xff0c;花费的时间也小了

13 l.append(p)14 p.start()15 for i inl :16 i.join()17 stop &＃61;time.time()18 print(&＃39;%s&＃39;%(stop-start))

I/O密集型

七、python标准模块----concurrent.futures