生产消费者模型(必须要理解并且牢记,默写内容):
from multiprocessing import Process,Queue import time,random,os def procducer(q): for i in range(10): res='包子%s' %i time.sleep(0.5) q.put(res) print('%s 生产了 %s' %(os.getpid(),res)) def consumer(q): while True: res=q.get() if res is None: break print('%s 吃 %s' %(os.getpid(),res)) time.sleep(random.randint(2,3)) if __name__ == '__main__': q=Queue() p=Process(target=procducer,args=(q,)) c=Process(target=consumer,args=(q,)) p.start() c.start() p.join() q.put(None) print('主')
今日概要:
1 守护进程vs守护线程（*）
2 互斥锁（**）
3 信号量（**）
4 生产者消费者模型（*****）
5 GIL（什么时候用进程，什么时候用线程）（*****） ===========>这里明天讲
守护进程和守护线程:
# #守护进程 # from multiprocessing import Process # import os,time,random # # def task(): # print('%s is running' %os.getpid()) # time.sleep(2) # print('%s is done' %os.getpid()) # # p = Process(target=time.sleep, args=(3,)) # # p.start() # # if __name__ == '__main__': # p=Process(target=task) # p.daemon = True #1、必须在p.start()之前 2：守护进程不能开启子进程 # p.start() # # p.join() # print('主') # # ''' # 举例说明守护进程的应用场景： # 假设有两个任务要干，要玩出并发的效果，使用进程的话可以让主进程 # 执行一个任务，然后开启一个子进程执行一个任务。 # # 如果这两个任务毫无关系，那么就像上面这么做就可以 # 如果主进程的任务在执行完毕后，子进程的任务没有存在的意义了 # 那么该子进程应该在开启之前就被设置成守护进程 # ''' #迷惑人的例子 #主进程代码运行完毕,守护进程就会结束 # from multiprocessing import Process # 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__': # p1=Process(target=foo) # p2 = Process(target=bar) # # p1.daemon=True # p1.start() # p2.start() # print("main-------") #打印该行则主进程代码结束,则守护进程p1应该被终止,可能会有p1任务执行的打印信息123,因为主进程打印main----时,p1也执行了,但是随即被终止 # # # #守护线程：等到该进程内所有非守护线程都运行完才死掉 # from multiprocessing import Process # from threading import Thread # import os,time,random # # def task(): # # t=Thread(target=time.sleep,args=(3,)) # # t.start() # print('%s is running' %os.getpid()) # time.sleep(2) # print('%s is done' %os.getpid()) # # if __name__ == '__main__': # t=Thread(target=task) # t.daemon = True #1、必须在t.start()之前 # t.start() # # t.join() # print('主') #迷惑人的例子 #主进程代码运行完毕,守护进程就会结束 from multiprocessing import Process 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.start() t2.start() print("main-------") ''' 123 456 main------- end123 end456 '''
理解补充:
这里的的意思就是:我们要开启守护进程,开启的方式要在start之前开启,然后用固定格式开启,
# 开启之后的结果就是,守护进程是等待主进程执行完了自己的代码之后就会挂掉,至于主进程里面的其他的子进程就不考虑了,
# 即便那些子进程还没有执行完成我们也不会去继续执行了,程序到主进程执行完成后就结束了,这里就是守护进程的作用.
# 至于他的应用场景,就是在实现并发的情况下,我们在主进程里面需要开启其他的进程来帮助我们执行任务,这些任务彼此之间是有关联的,
# 我们只需要在主进程执行完后就不需要其他的子进程继续去执行了,这个时候我们的子进程就需要被设置为守护进程
互斥锁--------进程:
from multiprocessing import Process,Lock import os,time,random def task(mutex): mutex.acquire() print('%s print 1' %os.getpid()) time.sleep(random.randint(1,3)) print('%s print 2' %os.getpid()) time.sleep(random.randint(1, 3)) print('%s print 3' %os.getpid()) mutex.release() if __name__ == '__main__': # p1=Process(target=task) # p2=Process(target=task) # p3=Process(target=task) # p1.start() # p1.join() # p2.start() # p2.join() # p3.start() # p3.join() mutex=Lock() p1=Process(target=task,args=(mutex,)) p2=Process(target=task,args=(mutex,)) p3=Process(target=task,args=(mutex,)) p1.start() p2.start() p3.start()
锁进程的时候,当主进程执行完毕后,守护进程立即挂掉,即便此时还有没有执行完了子进程也不会去执行了,整个程序立即结束
互斥锁-------------线程:
from threading import Thread,Lock import time n=100 def task(): # global n # mutex.acquire() # temp=n # time.sleep(0.1) # n=temp-1 # mutex.release() global n with mutex: temp=n time.sleep(0.1) n=temp-1 if __name__ == '__main__': mutex=Lock() t_l=[] for i in range(100): t=Thread(target=task) t_l.append(t) t.start() for t in t_l: t.join() print(n)
线程里面有两种情况,如果主线程执行完了此时守护线程还没有完,就会立即挂掉,但是,如果主线程已经完了,守护线程还没有完,那么它作为守护线程也会挂掉.
这里有一个例子,可以加深对于锁的理解:
from multiprocessing import Process,Lock import json import os import time import random # import json # with open('db.txt','w',encoding='utf-8') as f: # json.dump({'count':1},f) # 在程序运行前先运行一下这上面的三行代码 def search(): with open('db.txt',encoding='utf-8') as f: dic=json.load(f) print('%s 剩余票数 %s' %(os.getpid(),dic['count'])) def get(): with open('db.txt',encoding='utf-8') as read_f: dic=json.load(read_f) if dic['count'] > 0: dic['count']-=1 time.sleep(random.randint(1,3)) #模拟手速+网速 with open('db.txt','w',encoding='utf-8') as write_f: json.dump(dic,write_f) print('%s 抢票成功' %os.getpid()) def task(mutex): search() mutex.acquire() get() mutex.release() if __name__ == '__main__': # for i in range(20): # p=Process(target=task) # p.start() # p.join() mutex = Lock() for i in range(10): p = Process(target=task, args=(mutex, )) p.start()
信号量(理解即可,不必深究):
from multiprocessing import Process,Semaphore # from threading import Thread,Semaphore import time,random,os def task(sm): with sm: print('%s 上厕所' %os.getpid()) time.sleep(random.randint(1,3)) if __name__ == '__main__': sm=Semaphore(3) for i in range(10): p=Process(target=task,args=(sm,)) p.start() # 这里是信号量,它跟进程池就是多了一个锁的概念,资源抢占,相当于是在一个公司里有很多# 人要干活, # 每个人的分工不同,大家平日里都是各做各的事情,但是大家都会牵扯到使用打印机的情况,当大家都扎堆使用打印机的时候, # 这里就牵扯到了信号量的概念,打印机只有3个,但是使用它的人却源源不断理解到这即可
队列,堆栈,优先级:
from multiprocessing import Queue #进程队列 # q=Queue(3) # # q.put({'a':1}) # q.put('xxxxx') # q.put(3) # q.put(4) # print(q.get()) # print(q.get()) # print(q.get()) # print(q.get()) import queue #线程队列 #队列 # q=queue.Queue(3) # q.put({'a':1}) # q.put('xxxxx') # q.put(3) # q.put(4) # print(q.get()) # print(q.get()) # print(q.get()) # print(q.get()) #优先级队列 # q=queue.PriorityQueue(3) # q.put((10,{'a':1})) # q.put((-1,'xxxxx')) # q.put((0,3)) # # q.put(4) # # print(q.get()) # print(q.get()) # print(q.get()) # print(q.get()) #堆栈 # q=queue.LifoQueue(3) # q.put({'a':1}) # q.put('xxxxx') # q.put(3) # # q.put(4) # # print(q.get()) # print(q.get()) # print(q.get()) # print(q.get())
# #pip install requests
# import requests
# from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
# from threading import current_thread
# import time
# import os
#
# def get(url):
# print('%s GET %s' %(os.getpid(),url))
# response=requests.get(url)
# time.sleep(3)
# if response.status_code == 200:
# return {'url':url,'text':response.text}
#
# def parse(obj):
# res=obj.result()
# print('[%s] <%s> (%s)' % (os.getpid(), res['url'],len(res['text'])))
#
# if __name__ == '__main__':
# urls = [
# 'https://www.python.org',
# 'https://www.baidu.com',
# 'https://www.jd.com',
# 'https://www.tmall.com',
# ]
# # t=ThreadPoolExecutor(2)
# t=ProcessPoolExecutor(2)
# for url in urls:
# t.submit(get,url).add_done_callback(parse)
# t.shutdown(wait=True)
#
# print('主',os.getpid())
# '''
# 异步调用：
# 提交完任务（为该任务绑定一个回调函数），不用再原地等任务执行完毕拿到结果，可以直接提交下一个任务
# 一个任务一旦执行完毕就会自动触发回调函数的运行
#
# 回调函数的参数是单一的：
# 回调函数的参数就是它所绑定任务的返回值
#
# '''
#pip install requests
import requests
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from threading import current_thread
import time
import os
def get(url):
print('%s GET %s' %(current_thread().getName(),url))
response=requests.get(url)
time.sleep(3)
if response.status_code == 200:
return {'url':url,'text':response.text}
def parse(obj):
res=obj.result()
print('[%s] <%s> (%s)' % (current_thread().getName(), res['url'],len(res['text'])))
if __name__ == '__main__':
urls = [
'https://www.python.org',
'https://www.baidu.com',
'https://www.jd.com',
'https://www.tmall.com',
]
t=ThreadPoolExecutor(2)
for url in urls:
t.submit(get,url).add_done_callback(parse)
t.shutdown(wait=True)
print('主',os.getpid())