生产消费者模型(必须要理解并且牢记,默写内容):
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 requestsfrom concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorfrom threading import current_threadimport timeimport 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())
时间： 2024-10-12 02:34:52