四 线程死锁和递归锁

在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁，因为系统判断这部分资源都正在使用，所有这两个线程在无外力作用下将一直等待下去。下面是一个死锁的例子：

import threading,time

class myThread(threading.Thread):

def doA(self):

lockA.acquire()

print(self.name,"gotlockA",time.ctime())

time.sleep(3)

lockB.acquire()

print(self.name,"gotlockB",time.ctime())

lockB.release()

lockA.release()

def doB(self):

lockB.acquire()

print(self.name,"gotlockB",time.ctime())

time.sleep(2)

lockA.acquire()

print(self.name,"gotlockA",time.ctime())

lockA.release()

lockB.release()

def run(self):

self.doA()

self.doB()

if __name__=="__main__":

lockA=threading.Lock()

lockB=threading.Lock()

threads=[]

for i in range(5):

threads.append(myThread())

for t in threads:

t.start()

for t in threads:

t.join()#等待线程结束，后面再讲。

解决办法：使用递归锁

为了支持在同一线程中多次请求同一资源，python提供了“可重入锁”：threading.RLock。RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次acquire。直到一个线程所有的acquire都被release，其他的线程才能获得资源。

递归锁

import threading,time

class MyThread(threading.Thread):

def actionA(self):

r_lock.acquire()

print("gotA",self.name,time.ctime())

time.sleep(2)

r_lock.acquire()

print("gotB",self.name,time.ctime())

time.sleep(1)

r_lock.release()

r_lock.release()

def actionB(self):

r_lock.acquire()

print("gotC", self.name, time.ctime())

time.sleep(2)

r_lock.acquire()

print("gotD", self.name, time.ctime())

time.sleep(1)

r_lock.release()

r_lock.release()

def run(self):

self.actionA()

self.actionB()

if __name__ == '__main__':

r_lock = threading.RLock()

l = []

for i in range(5):

t = MyThread()

t.start()

l.append(t)

for i in l:

i.join()

print("ending...")

六 同步条件(Event)

条件同步和条件变量同步差不多意思，只是少了锁功能，因为条件同步设计于不访问共享资源的条件环境。event=threading.Event()：条件环境对象，初始值 为False；

event.isSet()：返回event的状态值；

event.wait()：如果 event.isSet()==False将阻塞线程；

event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；

event.clear()：恢复event的状态值为False。

实例1：

import threading,time

class Boss(threading.Thread):

def run(self):

print("BOSS：今晚大家都要加班到22:00。")

event.isSet() or event.set()

time.sleep(5)

print("BOSS：<22:00>可以下班了。")

event.isSet() or event.set()

class Worker(threading.Thread):

def run(self):

event.wait()

print("Worker：哎……命苦啊！")

time.sleep(0.25)

event.clear()

event.wait()

print("Worker：OhYeah!")

if __name__=="__main__":

event=threading.Event()

threads=[]

for i in range(5):

threads.append(Worker())

threads.append(Boss())

for t in threads:

t.start()

for t in threads:

t.join()

实例2：

import threading,time

import random

def light():

if not event.isSet():

event.set() #wait就不阻塞 #绿灯状态

count = 0

while True:

if count < 10:

print('\033[42;1m--green light on---\033[0m')

elif count <13:

print('\033[43;1m--yellow light on---\033[0m')

elif count <20:

if event.isSet():

event.clear()

print('\033[41;1m--red light on---\033[0m')

else:

count = 0

event.set() #打开绿灯

time.sleep(1)

count +=1

def car(n):

while 1:

time.sleep(random.randrange(10))

if event.isSet(): #绿灯

print("car [%s] is running.." % n)

else:

print("car [%s] is waiting for the red light.." %n)

if __name__ == '__main__':

event = threading.Event()

Light = threading.Thread(target=light)

Light.start()

for i in range(3):

t = threading.Thread(target=car,args=(i,))

t.start()

八 多线程利器(queue)

创建一个“队列”对象

import Queue

q = Queue.Queue(maxsize = 10)

Queue.Queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过Queue的构造函数的可选参数maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限。

将一个值放入队列中

q.put(10)

调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；第二个block为可选参数，默认为

1。如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0，put方法将引发Full异常。

将一个值从队列中取出

q.get()

调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。

Python Queue模块有三种队列及构造函数:

1、Python Queue模块的FIFO队列先进先出。 class queue.Queue(maxsize)

2、LIFO类似于堆，即先进后出。 class queue.LifoQueue(maxsize)

3、还有一种是优先级队列级别越低越先出来。 class queue.PriorityQueue(maxsize)

此包中的常用方法(q = Queue.Queue()):

q.qsize() 返回队列的大小

q.empty() 如果队列为空，返回True,反之False

q.full() 如果队列满了，返回True,反之False

q.full 与 maxsize 大小对应

q.get([block[, timeout]]) 获取队列，timeout等待时间

q.get_nowait() 相当q.get(False)

非阻塞 q.put(item) 写入队列，timeout等待时间

q.put_nowait(item) 相当q.put(item, False)

q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号

q.join() 实际上意味着等到队列为空，再执行别的操作

实例:

import time,random

import queue,threading

q = queue.Queue()

def Producer(name):

count = 0

while count <20:

time.sleep(random.randrange(3))

q.put(count)

print('Producer %s has produced %s baozi..' %(name, count))

count +=1

def Consumer(name):

count = 0

while count <20:

time.sleep(random.randrange(4))

if not q.empty():

data = q.get()

print(data)

print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' %(name, data))

else:

print("-----no baozi anymore----")

count +=1

p1 = threading.Thread(target=Producer, args=('A',))

c1 = threading.Thread(target=Consumer, args=('B',))

p1.start()

c1.start()

实例2:

#实现一个线程不断生成一个随机数到一个队列中(考虑使用Queue这个模块)

# 实现一个线程从上面的队列里面不断的取出奇数

# 实现另外一个线程从上面的队列里面不断取出偶数

import random,threading,time

from queue import Queue

#Producer thread

class Producer(threading.Thread):

def __init__(self, t_name, queue):

threading.Thread.__init__(self,name=t_name)

self.data=queue

def run(self):

for i in range(10): #随机产生10个数字 ，可以修改为任意大小

randomnum=random.randint(1,99)

print ("%s: %s is producing %d to the queue!" % (time.ctime(), self.getName(), randomnum))

self.data.put(randomnum) #将数据依次存入队列

time.sleep(1)

print ("%s: %s finished!" %(time.ctime(), self.getName()))

#Consumer thread

class Consumer_even(threading.Thread):

def __init__(self,t_name,queue):

threading.Thread.__init__(self,name=t_name)

self.data=queue

def run(self):

while 1:

try:

val_even = self.data.get(1,5) #get(self, block=True, timeout=None) ,1就是阻塞等待,5是超时5秒

if val_even%2==0:

print ("%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(),self.getName(),val_even))

time.sleep(2)

else:

self.data.put(val_even)

time.sleep(2)

except: #等待输入，超过5秒 就报异常

print ("%s: %s finished!" %(time.ctime(),self.getName()))

break

class Consumer_odd(threading.Thread):

def __init__(self,t_name,queue):

threading.Thread.__init__(self, name=t_name)

self.data=queue

def run(self):

while 1:

try:

val_odd = self.data.get(1,5)

if val_odd%2!=0:

print ("%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(), self.getName(), val_odd))

time.sleep(2)

else:

self.data.put(val_odd)

time.sleep(2)

except:

print ("%s: %s finished!" % (time.ctime(), self.getName()))

break

#Main thread

def main():

queue = Queue()

producer = Producer('Pro.', queue)

consumer_even = Consumer_even('Con_even.', queue)

consumer_odd = Consumer_odd('Con_odd.',queue)

producer.start()

consumer_even.start()

consumer_odd.start()

producer.join()

consumer_even.join()

consumer_odd.join()

print ('All threads terminate!')

if __name__ == '__main__':

main()

一 进程的调用

调用方式1

from multiprocessing import Process

import time

def f(name):

time.sleep(1)

print('hello', name,time.ctime())

if __name__ == '__main__':

p_list=[]

for i in range(3):

p = Process(target=f, args=('alvin',))

p_list.append(p)

p.start()

for i in p_list:

p.join()

print('end')

调用方式2

from multiprocessing import Process

import time

class MyProcess(Process):

def __init__(self):

super(MyProcess, self).__init__()

#self.name = name

def run(self):

time.sleep(1)

print ('hello', self.name,time.ctime())

if __name__ == '__main__':

p_list=[]

for i in range(3):

p = MyProcess()

p.start()

p_list.append(p)

for p in p_list:

p.join()

print('end')

To show the individual process IDs involved, here is an expanded example:

from multiprocessing import Process

import os

import time

def info(title):

print("title:",title)

print('parent process:', os.getppid())

print('process id:', os.getpid())

def f(name):

info('function f')

print('hello', name)

if __name__ == '__main__':

info('main process line')

time.sleep(1)

print("------------------")

p = Process(target=info, args=('yuan',))

p.start()

p.join()

二 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：进程号。

import time

from multiprocessing import Process

def foo(i):

time.sleep(1)

print (p.is_alive(),i,p.pid)

time.sleep(1)

if __name__ == '__main__':

p_list=[]

for i in range(10):

p = Process(target=foo, args=(i,))

#p.daemon=True

p_list.append(p)

for p in p_list:

p.start()

# for p in p_list:

# p.join()

print('main process end')

三 进程间通讯

3.1 进程对列Queue

from multiprocessing import Process, Queue

import queue

def f(q,n):

#q.put([123, 456, 'hello'])

q.put(n*n+1)

print("son process",id(q))

if __name__ == '__main__':

q = Queue() #try: q=queue.Queue()

print("main process",id(q))

for i in range(3):

p = Process(target=f, args=(q,i))

p.start()

print(q.get())

print(q.get())

print(q.get())

3.2 管道

from multiprocessing import Process,Pipe

def foo(conn):

conn.send("ni hao ma")

print(conn.recv())

conn.close()

if __name__ == '__main__':

p_conn,c_conn = Pipe()

p = Process(target=foo,args=(c_conn,))

p.start()

print(p_conn.recv())

p_conn.send("hao ")

p_conn.close()

p.join()

3.3 Managers

from multiprocessing import Process,Manager

def foo(d,l,n):

d[n] = "1"

d["2"] = 2

l.append(n)

if __name__ == '__main__':

with Manager() as manager:

d = manager.dict() #{}

l = manager.list(range(3)) #[0,1,2]

p_li = []

for i in range(10):

p = Process(target=foo,args=(d,l,i))

p.start()

p_li.append(p)

for t in p_li:

t.join()

print(d)

print(l)

四 进程同步

from multiprocessing import Process,Lock

def foo(lock,i):

lock.acquire()

print("hello %s"%i)

lock.release()

if __name__ == '__main__':

lock = Lock()

for i in range(10):

Process(target=foo,args=(lock,i)).start()

五 进程池

from multiprocessing import Pool

import time,os

def foo(i):

time.sleep(1)

print(i)

print("son ",os.getpid())

return "HELLO %s"%i

def bar(arg):

print(arg)

print("bar ",os.getpid())

if __name__ == '__main__':

pool = Pool()

print("main ",os.getpid())

for i in range(100):

# 回调函数：就是某个动作或者函数执行成功后再去执行函数

pool.apply_async(func=foo,args=(i,),callback=bar)

pool.close()

pool.join() #join 与close调用顺序是固定的

print("end...")

六 协程

Greenlet

from greenlet import greenlet

def test1():

print(1)

gr2.switch()

print(4)

def test2():

print(2)

gr1.switch()

print(3)

gr1.switch()

if __name__ == '__main__':

gr1 = greenlet(test1)

gr2 = greenlet(test2)

gr2.switch()

Gevent

import gevent

import requests,time

start=time.time()

def f(url):

print('GET: %s' % url)

resp =requests.get(url)

data = resp.text

print('%d bytes received from %s.' % (len(data), url))

gevent.joinall([

gevent.spawn(f, 'https://www.python.org/'),

gevent.spawn(f, 'https://www.yahoo.com/'),

gevent.spawn(f, 'https://www.baidu.com/'),

gevent.spawn(f, 'https://www.sina.com.cn/'),

])

# f('https://www.python.org/')

#

# f('https://www.yahoo.com/')

#

# f('https://baidu.com/')

#

# f('https://www.sina.com.cn/')

print("cost time:",time.time()-start)