首页 > 代码库 > python--同步锁/递归锁/协程
python--同步锁/递归锁/协程
同步锁/递归锁/协程
1 同步锁
锁通常被用来实现对共享资源的同步访问，为每一个共享资源创建一个Lock对象，当你需需要访问该资源时，调用acquire（）方法来获取锁对象（如果其他线程已经获得了该锁，则当前线程需要等待其被释放），待资源访问完后，在调用release方式释放锁：
import threading
import time
def subnum():
global num
# num-=1
lock.acquire() #对用户进行加锁处理
#加锁只对用户数据 等第一个释放完之后才能执行下一个
temp=num
time.sleep(0.01)
num=temp-1 #对此公共变量进行-1操作
lock.release() #然后进程释放锁
num=100
l=[]
lock=threading.Lock()
for i in range(100):
t=threading.Thread(target=subnum)
t.start()
l.append(t)
for t in l:
t.join()
print("result:",num)
执行结果：
result: 0
2 死锁
所谓死锁：是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程成为死锁进程。
import threading
import time
mutexA=threading.Lock()
mutexB=threading.Lock()
class MyThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
self.fun1()
self.fun2()
def fun1(self):
mutexA.acquire()
print("I am %s,get res:%s---%s"%(self.name,"ResA",time.time()))
mutexB.acquire()
print("I am %s,get res:%s---%s"%(self.name,"ResB",time.time()))
mutexB.release()
mutexA.release()
def fun2(self):
mutexB.acquire()
print("I am %s,get res:%s---%s"%(self.name,"ResB",time.time()))
time.sleep(0.2)
mutexA.acquire()
print("I am %s,get res:%s---%s"%(self.name,"ResB",time.time()))
mutexA.release()
mutexA.release()
if __name__ == ‘__main__‘:
print("start------------%s",time.time())
for i in range(0,10):
my_thread=MyThread()
my_thread.start()
执行结果：
start------------%s 1494311688.1136441
I am Thread-1,get res:ResA---1494311688.1136441
I am Thread-1,get res:ResB---1494311688.1136441
I am Thread-1,get res:ResB---1494311688.1136441
I am Thread-2,get res:ResA---1494311688.1136441
3 递归锁
在python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。
import threading
import time
Rlock=threading.RLock()
class MyThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
self.fun1()
self.fun2()
def fun1(self):
Rlock.acquire() #如果锁被占用，则阻塞在这里，等待锁的释放
print("I am %s,get res:%s---%s"%(self.name,"ResA",time.time()))
Rlock.acquire() #count=2
print("I am %s,get res:%s---%s"%(self.name,"ResB",time.time()))
Rlock.release() #count-1
Rlock.release() #count-1=0
def fun2(self):
Rlock.acquire() #count=1
print("I am %s,get res:%s---%s"%(self.name,"ResB",time.time()))
time.sleep(0.2)
Rlock.acquire() #count=2
print("I am %s,get res:%s---%s"%(self.name,"ResA",time.time()))
Rlock.release() #coun-1
Rlock.release() #count-1
if __name__ == ‘__main__‘:
print("start-----------%s"%time.time())
for i in range(0,10):
my_thread=MyThread()
my_thread.start()
4 Event对象
线程的一个关键特性是每个线程都是独立运行且状态不可预测。我们需要使用threading库中的Event对象，对象包含一个可由线程设置的信号标志，它允许线程等待某些事情的发生，在初始情况下，Event对象中的信号标志被设置为假。如果有线程等待一个Event对象，而这个Event对象的标志位假，那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真，它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象，那么它将忽略这个事情，继续执行。
event.isSet()：返回event的状态值； event.wait()：如果 event.isSet()==False将阻塞线程； event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度； event.clear()：恢复event的状态值为False。
例如，我们有多个线程从Redis队列中读取数据来处理，这些线程都要尝试去链接Redis的服务，一般情况下，如果Redis连接不成功，这各个线程的代码中，都会去尝试重新连接，那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作：主线程中会去尝试连接Redis服务，如果正常的话，触发事件，各工作线程会尝试连接Redis服务。
#event 就是线程线程之间通信使用的
#event wait 默认值为False 就会被阻塞
#set 改成Ture 就是向下执行
import threading
import time
import logging
logging.basicConfig(level=logging.DEBUG,format="(%(threadName)-10s)%(message)s",)
def worker(event):
logging.debug("waiting for redis ready...")
event.wait() #if flag=False 阻塞，等待flag=true继续执行
logging.debug(‘redis ready,and connect to redis server and do some work[%s]‘,time.time())
time.sleep(1)
def main():
readis_ready=threading.Event() #flga=False
t1=threading.Thread(target=worker,args=(readis_ready,),name="t1")
t1.start()
t2=threading.Thread(target=worker,args=(readis_ready,),name="t2")
t2.start()
logging.debug("first of all,check redis server,make sure it is ok,and then trigger the redis teady event")
time.sleep(3) #simulate the check propress
readis_ready.set()
if __name__ == ‘__main__‘:
main()
执行结果：
(t1 )waiting for redis ready...
(t2 )waiting for redis ready...
(MainThread)first of all,check redis server,make sure it is ok,and then trigger the redis teady event
(t1 )redis ready,and connect to redis server and do some work[1494314141.0479438]
(t2 )redis ready,and connect to redis server and do some work[1494314141.0479438]
threading.Event的wait方法还接受了一个超时参数，默认情况下如果事件一致没有发生，wait方法会一直阻塞下去，而加入这个超时参数之后，如果阻塞时间超过这个参数设定的值之后，wait方法会返回。对应于上面的应用场景，如果Redis服务器一致没有启动，我们希望子线程能够打印一些日志来不断地提醒我们当前没有一个可以连接的Redis服务，我们就可以通过设置这个超时参数来达成这样的目的。
import threading
import time
import logging
logging.basicConfig(level=logging.DEBUG,format="(%(threadName)-10s)%(message)s",)
def worker(event):
logging.debug("waiting for redis ready....")
while not event.is_set():#现在是false的时候
event.wait(3) #每隔三秒钟提示一下 #一直打印
logging.debug("wait........")
logging.debug("redis ready,and connect to redis server and do mome work [%s]",time.ctime())
time.sleep(1)
def main():
readis_ready=threading.Event() #flga=False
t1=threading.Thread(target=worker,args=(readis_ready,),name="t1")
t1.start()
t2=threading.Thread(target=worker,args=(readis_ready,),name="t2")
t2.start()
logging.debug("first of all,check redis server,make sure it is ok,and then trigger the redis teady event")
time.sleep(3) #simulate the check propress
readis_ready.set()
if __name__ == ‘__main__‘:
main()
执行结果
(t1 )waiting for redis ready....
(t2 )waiting for redis ready....
(MainThread)first of all,check redis server,make sure it is ok,and then trigger the redis teady event
(t1 )wait........
(t1 )redis ready,and connect to redis server and do mome work [Tue May 9 16:56:18 2017]
(t2 )wait........
(t2 )redis ready,and connect to redis server and do mome work [Tue May 9 16:56:18 2017]
5 信号量
Semaphore管理一个内置的计数器：
每当调用acquire()时内置计数器-1
调用release()时内置计数器+1
基数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。
实例：（同时只有5个线程可以获得semaphore，即可以限制最大连接数为5）
import threading
import time
semaphore=threading.Semaphore(5)
def func():
semaphore.acquire() #相当于加了一把锁 同时跑五个线程
print(threading.currentThread().getName()+"get semaphore")
time.sleep(3) #等待3秒后
semaphore.release() #五个线程同时释放掉
for i in range(20):
t1=threading.Thread(target=func)
t1.start()
6 multiprocessing模块
由于GIL的存在，python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。
Multiprocessing包是python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在python程序内部编写的函数。该Process对象与Thread对象的用法相同，也有start(),run(),join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition类（这些对象可以像多线程那样，
普通通过参数传递给各个进程），用以同步进程，其用法与threading包中的同名类一致。所以，multiprocessing的很大一部分与threading使用同一套API，只不过换到了多进程的情境。
Python的进程调用：
Process类调用
from multiprocessing import Process
import time
def f(name):
print("hello",name,time.ctime())
time.sleep(5)
if __name__ == ‘__main__‘:
p_l=[]
for i in range(3):
p=Process(target=f,args=("alvin:%s"%i,))
p_l.append(p)
p.start()
for p in p_l:
p.join()
print("end")
执行结果：
hello alvin:0 Tue May 9 17:49:10 2017
hello alvin:1 Tue May 9 17:49:10 2017
hello alvin:2 Tue May 9 17:49:10 2017
end
继承Process类调用
from multiprocessing import Process
import time
class MyProcess(Process):
def __init__(self):
super().__init__()
def run(self):
print("hello",self.name,time.ctime())
time.sleep(1)
if __name__ == ‘__main__‘:
p_l=[]
for i in range(3):
p=MyProcess()
p.start()
p_l.append(p)
for p in p_l:
p.join()
print("end")
Process类
构造方法：
Process([group [, target [, name [, args [, kwargs]]]]])
group: 线程组，目前还没有实现，库引用中提示必须是None；
target: 要执行的方法；
name: 进程名；
args/kwargs: 要传入方法的参数。
实例方法：
is_alive()：返回进程是否在运行。
join([timeout])：阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。
start()：进程准备就绪，等待CPU调度
run()：strat()调用run方法，如果实例进程时未制定传入target，这star执行t默认run()方法。
terminate()：不管任务是否完成，立即停止工作进程
属性：
daemon：和线程的setDeamon功能一样
name：进程名字。
pid：进程号。
from multiprocessing import Process
import os
import time
def info(name):
print("name:",name)
print("parent process:",os.getppid())
print("process id:",os.getpid())
print("-----------")
time.sleep(2)
def foo(name):
info(name)
if __name__ == ‘__main__‘:
info("main process line")
p1=Process(target=info,args=("alvin",))
p2=Process(target=info,args=("egon",))
p1.start()
p2.start()
p1.join()
p2.join()
print("ending")
执行结果：
name: main process line
parent process: 8032
process id: 9400
-----------
name: alvin
parent process: 9400
process id: 9252
-----------
name: egon
parent process: 9400
process id: 10556
-----------
ending
通过tasklist检测每一个进程号（PID）对应的进程名
7 协程
优点：1 由于单线程，不能再切换
2 不再有任何锁的概念
yield与携程：产生了并发
yeild与携程
import time
def consumer():#消费者的
r=""
while True:
#3 consumer通过yield拿到消息，处理，又通过yield把结果传回
n=yield r
if not n:
return
print("[CONSUMER]--Consuming %s...."%n)
time.sleep(1)
r="200 OK"
def produce(c):#消费者
next(c) #取生成对象的一个值
n=0
if __name__ == ‘__main__‘:
while n<5:
n=n+1
print("[PRODUCER]--Producing %s..."%n)
#2 然后，一旦产生了东西，通过c.send（n）切换到consumer执行
cr=c.send(n)
#4 produce拿到consumer处理的结果，继续生产下一条消息
print("[PRODUCER]Consumer return:%s"%cr)
#5 produce决定不生产了，通过c.close()关闭consumer，整个过程结束
c.close()
if __name__ == ‘__main__‘:
c=consumer() #产生一个生成器
produce(c)
执行结果：
[PRODUCER]--Producing 1...
[CONSUMER]--Consuming 1....
[PRODUCER]Consumer return:200 OK
[PRODUCER]--Producing 2...
[CONSUMER]--Consuming 2....
[PRODUCER]Consumer return:200 OK
[PRODUCER]--Producing 3...
[CONSUMER]--Consuming 3....
[PRODUCER]Consumer return:200 OK
[PRODUCER]--Producing 4...
[CONSUMER]--Consuming 4....
[PRODUCER]Consumer return:200 OK
[PRODUCER]--Producing 5...
[CONSUMER]--Consuming 5....
[PRODUCER]Consumer return:200 OK
Greenlet模块 是协程的基础
Greenlet机制的主要思想是：生成器函数或者协程函数中的yield语句挂起函数的执行，greenlet是python中实现我们所谓的”Coroutine（携程）”的一个基础库
from greenlet import greenlet
def test1():
print (12)
gr2.switch()
print (34)
gr2.switch()
def test2():
print (56)
gr1.switch()
print (78)
gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()
基于greenlet的框架
Gevent模块实现携程
Python通过yield提供了对协程的基本支持，但是不完全，而第三方的gevent为python提供了比较完成的协程支持
当一个greenlet遇到IO操作时，比如访问网络，就自动切换到其他的greenlet，等到IO操作完成，再在适当的时候切换回来继续执行，由于IO操作非常耗时，经常使程偶greenlet序处于等待状态，有了gevent为我们自动切换协程，就保证greenlet在运行，而不是等待IO。
import gevent
import time
def foo():
print("running in foo")
gevent.sleep(2)
print("switch to foo again")
def bar():
print("switch to bar")
gevent.sleep(5)
print("switch to bar again")
start=time.time()
gevent.joinall(
[gevent.spawn(foo),
gevent.spawn(bar)]
)
print(time.time()-start)
当然，实际代码里，我们不会用gevent.sleep()去切换协程，而是在执行到IO操作时，gevent自动切换，代码如下：
from gevent import monkey
monkey.patch_all()
import gevent
from urllib import request
import time
def f(url):
print(‘GET: %s‘ % url)
resp = request.urlopen(url)
data = http://www.mamicode.com/resp.read()>
python--同步锁/递归锁/协程