一、 互斥锁(Lock)与递归锁(RLock)机制

1.1 由互斥锁(Lock)产生的死锁现象：

#互斥锁(死锁现象)：

#死锁现象：

from threading importLock

lock=Lock()

lock.acquire()print(123)

lock.acquire()#等待获取锁(死锁状态)

print(456)

lock.release()#等待释放锁

lock.release()print('程序结束！！！')#科学家吃面产生的死锁现象：

from threading importLock, Threadimporttime

noodle_lock= Lock() #面条锁

fork_lock = Lock() #叉子锁

defeat1(name):

noodle_lock.acquire()

time.sleep(0.5)print('%s拿到面了' %name)

fork_lock.acquire()print('%s拿到叉子了' %name)print('%s吃面' %name)

fork_lock.release()print('%s放下叉子了' %name)

noodle_lock.release()print('%s放下面了' %name)defeat2(name):

fork_lock.acquire()print('%s拿到叉子了' %name)

noodle_lock.acquire()print('%s拿到面了' %name)print('%s吃面' %name)

noodle_lock.release()print('%s放下面了' %name)

fork_lock.release()print('%s放下叉子了' %name)#三个线程(科学家)需同时获取叉子和面条才能吃到面条

Thread(target=eat1, args=('Annie',)).start()

Thread(target=eat2, args=('Lisa',)).start()

Thread(target=eat1, args=('Jane',)).start()

View Code

1.2 解决由互斥锁产生的死锁，可重入锁(递归锁：RLock)解决方法：

#递归锁(可重入锁)：

#方式一：#RLock：解决死锁问题(建议少用递归锁，一般程序的递归锁的出现都是程序设计不合理导致)：

from threading importRLock, Threadimporttime

fork_lock= noodle_lock =RLock()defeat1(name):

noodle_lock.acquire()#获取递归锁(进入第一层)

print('%s拿到面了' %name)

time.sleep(0.5)

fork_lock.acquire()#获取递归锁(进入第二层)

print('%s拿到叉子了' %name)print('%s吃面' %name)

fork_lock.release()#释放递归锁(出第二层)

print('%s放下叉子了' %name)

noodle_lock.release()#释放递归锁(出第一层)

print('%s放下面了' %name)defeat2(name):

fork_lock.acquire()print('%s拿到叉子了' %name)

noodle_lock.acquire()print('%s拿到面了' %name)print('%s吃面' %name)

noodle_lock.release()print('%s放下面了' %name)

fork_lock.release()print('%s放下叉子了' %name)

Thread(target=eat1, args=('Annie',)).start() #线程一

Thread(target=eat2, args=('Lisa',)).start() #线程二

Thread(target=eat1, args=('Jane',)).start() #线程三

#方式二：#直接使用互斥锁结果科学家吃面问题：

from threading importLock, Thread

lock=Lock()defeat1(name):

lock.acquire()#获取锁

print('%s拿到面了' %name)print('%s拿到叉子了' %name)print('%s吃面' %name)print('%s放下叉子了' %name)print('%s放下面了' %name)

lock.release()#释放锁

defeat2(name):

lock.acquire()print('%s拿到叉子了' %name)print('%s拿到面了' %name)print('%s吃面' %name)print('%s放下面了' %name)print('%s放下叉子了' %name)

lock.release()

Thread(target=eat1, args=('Annie',)).start() #线程一

Thread(target=eat2, args=('Lisa',)).start() #线程二

Thread(target=eat1, args=('Jane',)).start() #线程三

View Code

1.3 总结：

gil锁机制：保证线程同一时刻只能一个线程访问CPU,不可能有两个线程同时在CPU上执行指令

lock锁机制：保证某一段代码 在没有执行完毕之后,不可能有另一个线程也执行这一段代码

二、 线程锁实现

为什么要使用线程锁：

1. 同一时刻同一段代码,只能有一个进程来执行这段代码

2. 锁的应用场景,当多个进程需要操作同一个文件/数据库的时候 ,

3. 会产生数据不安全,我们应该使用锁来避免多个进程同时修改一个文件

示例：

#示例：

importjsonimporttimefrom multiprocessing importProcess, Lock#查询余票

defsearch_ticket(name):

with open('ticket', encoding='utf-8') as f:

dic=json.load(f)print('%s查询余票为%s' % (name, dic['count']))#购买车票

defbuy_ticket(name):

with open('ticket', encoding='utf-8') as f:

dic=json.load(f)

time.sleep(2)if dic['count'] >= 1:print('%s买到票了' %name)

dic['count'] -= 1time.sleep(2)

with open('ticket', mode='w', encoding='utf-8') as f:

json.dump(dic, f)else:print('余票为0,%s没买到票' %name)#使用线程锁方式一：

defuse1(name, lock):

search_ticket(name)print('%s在等待' %name)

lock.acquire()#获取锁

print('%s开始执行了' %name)

buy_ticket(name)

lock.release()#释放锁

#使用线程锁方式二：

defuse2(name, lock):"""# with lock:

# 代码块

# 上下文管理：在__enter__方法中获取锁(acquire)，在__exit__方法中释放锁(release)"""search_ticket(name)print('%s在等待' %name)

with lock:#获取锁 + 释放锁

print('%s开始执行了' %name)

buy_ticket(name)if __name__ == '__main__':

lock=Lock()

l= ['alex', 'wusir', 'baoyuan', 'taibai']for name inl:

Process(target=use1, args=(name, lock)).start() #方式一

Process(target=use2, args=(name, lock)).start() #方式二

View Code

三、 队列(Queue，LifoQueue，PriorityQueue)的使用

3.1 先进先出队列(Queue):

#先进先出队列#Queue就是一个线程队列的类,自带lock锁,实现了线程安全的数据类型

from queue importQueue

q=Queue()

q.put({1, 2, 3})

q.put_nowait('abc')print(q.get_nowait()) #获取一个值，如果没有抛出异常

print(q.get())

q.empty()#判断是否为空

q.full() #判断是否为满

q.qsize() #查看队列的大小

View Code

3.2 先进后出队列(栈：LifoQueue):

#先进后出的队列(last in first out):#线程安全的队列 栈和后进先出的场景都可以用

from queue importLifoQueue

lfq=LifoQueue()

lfq.put(1)

lfq.put('abc')

lfq.put({'1', '2'})print(lfq.get()) #{'2', '1'}

print(lfq.get()) #abc

print(lfq.get()) #1

View Code

3.3 优先级队列(PriorityQueue):

#优先级队列:

from queue importPriorityQueue

pq=PriorityQueue()

pq.put((10, 'aaa'))

pq.put((2, 'bbb'))

pq.put((20, 'ccc'))print(pq.get()) #最想获取到优先级最高的2,以元组形式返回 (2, 'bbb')

print(pq.get()) #(10, 'aaa')

print(pq.get()) #(20, 'ccc')

View Code

四、线程队列(生产者与消费者模型)

#生产者与消费者示例：

importtimeimportrandomfrom queue importQueuefrom threading importThread#生产者

defproducer(q):for i in range(10):

time.sleep(random.random())

food= 'Spam %s' %iprint('%s生产了%s' % ('Jane', food))

q.put(food)#消费者

defconsumer(q, name):whileTrue:

food= q.get() #food = 食物/None

if not food: break #当消费者消费完成后，最后拿到None时，退出消费者程序

time.sleep(random.uniform(1, 2))print('%s 吃了 %s' %(name, food))if __name__ == '__main__':

q=Queue()

p1= Thread(target=producer, args=(q,)) #生产者

p1.start()

c1= Thread(target=consumer, args=(q, 'Lisa')) #消费者

c1.start()

c2= Thread(target=consumer, args=(q, 'Annie')) #消费者

c2.start()

p1.join()

q.put(None)#生产者完成生产后，队列最后加入None，表示已经生产完成

q.put(None) #每个消费者需要None退出程序

View Code

五、守护线程：

5.1 守护线程的定义：

1. 主线程会等待子线程的结束而结束

2. 守护线程会守护主线程和所有的子线程

3. 守护线程会随着主线程的结束而结束，主线程结束，进程资源回收，守护线程被销毁

示例：

#守护线程示例：

importtimefrom threading importThread#守护线程

defdaemon_func():whileTrue:

time.sleep(0.5)print('守护线程')#其他子线程

defson_func():print('start son')

time.sleep(5)print('end son')

t= Thread(target=daemon_func) #开启守护线程

t.daemon =True

t.start()

Thread(target=son_func).start() #开启son_func子线程

time.sleep(3)print('主线程结束') #主线程代码结束后，等待子线程son_func结束

View Code

总结：

守护进程 :只会守护到主进程的代码结束

守护线程 :会守护所有其他非守护线程的结束

六、线程池

线程池模块与进程池共用同一个模块：concureent.futures

示例：

#线程池：

from urllib.request importurlopenfrom concurrent.futures import ThreadPoolExecutor #导入线程池类

#获取网页

defget_html(name, addr):

ret=urlopen(addr)return {'name': name, 'content': ret.read()}#保存数据

defcache_page(ret_obj):

dic=ret_obj.result()

with open(dic['name'] + '.html', 'wb') as f:

f.write(dic['content'])

url_dic={'协程': 'http://www.cnblogs.com/Eva-J/articles/8324673.html','线程': 'http://www.cnblogs.com/Eva-J/articles/8306047.html','目录': 'https://www.cnblogs.com/Eva-J/p/7277026.html','百度': 'http://www.baidu.com','sogou': 'http://www.sogou.com'}#创建20个线程#方式一：

t = ThreadPoolExecutor(20) #实例化线程池对象

for url inurl_dic:

task= t.submit(get_html, url, url_dic[url]) #提交线程任务

task.add_done_callback(cache_page) #函数回调，数据保存

#方式二：

with ThreadPoolExecutor(20) as t: #上下文管理方式实现实例化的线程池

for url inurl_dic:

task=t.submit(get_html, url, url_dic[url])

task.add_done_callback(cache_page)

View Code