开发者

Python全栈之队列详解

开发者 https://www.devze.com 2022-12-07 11:19 出处:网络 作者: 熬夜泡枸杞
目录1. lock互斥锁2. 事件_红绿灯效果2.1 信号量_semaphore2.2 事件_红绿灯效果3. queue进程队列4. 生产者消费者模型5. joinablequeue队列使用6. 总结1. lock互斥锁
目录
  • 1. lock互斥锁
  • 2. 事件_红绿灯效果
    • 2.1 信号量_semaphore
    • 2.2 事件_红绿灯效果
  • 3. queue进程队列
    • 4. 生产者消费者模型
      • 5. joinablequeue队列使用
        • 6. 总结

          1. lock互斥锁

          知识点:

          lock.acquire()# 上锁
          lock.release()# 解锁
          #同一时间允许一个进程上一把锁 就是Lock
          	加锁可以保证多个进程修改同一块数据时,同一时间只能有一个任务可以进行修改,即串行的修改,没错,速度是慢了,但牺牲速度却保证了数据安全。
          #同一时间允许多个进程上多把锁 就是[信号量Semaphore]
          	信号量是锁的变形: 实际实现是 计数器 + 锁,同时允许多个进程上锁	
          # 互斥锁Lock : 互斥锁就是进程的互相排斥,谁先抢到资源,谁就上锁改资源内容,为了保证数据的同步性
          # 注意:多个锁一起上,不开锁,会造成死锁.上锁和解锁是一对.
          

          程序实现:

          # ### 锁 lock 互斥锁
          from multiprocessing import Process,Lock
          """ 上锁和解锁是一对, 连续上锁不解锁是死锁 ,只有在解锁的状态下,其他进程才有机会上锁 """
          """
          # 创建一把锁
          lock = Lock()
          # 上锁
          lock.acquire()
          # lock.acquire() # 连续上锁,造成了死锁现象;
          print("我在袅袅炊烟 ..  你在焦急等待 ... 厕所进行时 ... ")
          # 解锁
          lock.release()
          """
          # ### 12306 抢票软件
          import json,time,random
          # 1.读写数据库当中的票数
          def wr_info(sign , dic=None):
          	if sign == "r":
          		with open("ticket",mode="r",encoding="utf-8") as fp:
          			dic = json.load(fp)
          		return dic
          	elif sign == "w":
          		with open("ticket",mode="w",encoding="utf-8") as fp:
          			json.dump(dic,fp)
          # dic = wr_info("w",dic={"count":0})
          # print(dic , type(dic) )
          # 2.执行抢票的方法
          def get_ticket(person):
          	# 先获取数据库中实际票数
          	dic = wr_info("r")
          	# 模拟一下网络延迟
          	time.sleep(random.uniform(0.1,0.7))
          	# 判断票数
          	if dic["count"] > 0:
          		print("{}抢到票了".format(person))
          		# 抢到票后,让当前票数减1
          		dic["count"] -= 1
          		# 更新数据库中的票数
          		wr_info("w",dic)
          	else:
          		print("{}没有抢到票哦".format(person))
          # 3.对抢票和读写票数做一个统一的调用
          def main(person,lock):
          	# 查看剩余票数
          	dic = wr_info("r")
          	print("{}查看票数剩余: {}".format(person,dic["count"]))
          	# 上锁
          	lock.acquire()
          	# 开始抢票
          	get_ticket(person)
          	# 解锁 
          	lock.release()
          if __name__ == "__main__":
          	lock = Lock()
          	lst = ["梁新宇","康裕康","张保张","于朝志","薛宇健","韩瑞瑞","假摔先","刘子涛","黎明辉","赵凤勇"]
          	for i in lst:
          		p = Process(    target=main,args=(  i  , lock  )   )
          		p.start()
          """
          创建进程,开始抢票是异步并发程序
          直到开始抢票的时候,变成同步程序,
          先抢到锁资源的先执行,后抢到锁资源的后执行;
          按照顺序依次执行;是同步程序;
          抢票的时候,变成同mvrxQ步程序,好处是可以等到数据修改完成之后,在让下一个人抢,保证数据不乱。
          如果不上锁的话,只剩一张票的时候,那么所有的人都能抢到票,因为程序执行的速度太快,所以接近同步进程,导致数据也不对。
          """	
          

          ticket文件

          {"count": 0}
          

          Python全栈之队列详解

          2. 事件_红绿灯效果

          2.1 信号量_semaphore

          # ### 信号量 Semaphore 本质上就是锁,只不过是多个进程上多把锁,可以控制上锁的数量
          """Semaphore = lock + 数量 """
          from multiprocessing import Semaphore , Process
          import time , random
          """
          	# 同一时间允许多个进程上5把锁
          	sem = Semaphore(5)
          	#上锁
          	sem.acquire()
          	print("执行操作 ... ")
          	#解锁
          	sem.release()
          """
          def singsong_ktv(person,sem):
          	# 上锁
          	sem.acquire()
          	print("{}进入了唱吧ktv , 正在唱歌 ~".format(person))
          	# 唱一段时间
          	time.sleep( random.randrange(4,8) ) # 4 5 6 7
          	print("{}离开了唱吧ktv , 唱完了 ... ".format(person))
          	# 解锁
          	sem.release()
          if __name__ == "__main__":
          	sem = Semaphore(5)
          	lst = ["赵凤勇" , "沈思雨", "赵万里" , "张宇" , "假率先" , "孙杰龙" , "陈璐" , "王雨涵" , "杨元涛" , "刘一凤"   ]
          	for i  in lst:
          		p = Process(target=singsong_ktv , args = (i , sem)		)
          		p.start()
          """
          # 总结: Semaphore 可以设置上锁的数量 , 同一时间上多把锁
          创建进程时,是异步并发,执行任务时,是同步程序;
          """
          # 赵万里进入了唱吧ktv , 正在唱歌 ~
          # 赵凤勇进入了唱吧ktv , 正在唱歌 ~
          # 张宇进入了唱吧ktv , 正在唱歌 ~
          # 沈思雨进入了唱吧ktv , 正在唱歌 ~
          # 孙杰龙进入了唱吧ktv , 正在唱歌 ~
          

          2.2 事件_红绿灯效果

          # ### 事件 (Event)
          """
          # 阻塞事件 :
          	e = Event()生成事件对象e   
          	e.wait()动态给程序加阻塞 , 程序当中是否加阻塞完全取决于该对象中的is_set() [默认返回值是False]
              # 如果是True  不加阻塞
              # 如果是False 加阻塞
          # 控制这个属性的值
              # set()方法     将这个属性的值改成True
              # clear()方法   将这个属性的值改成False
              # is_set()方法  判断当前的属性是否为True  (默认上来是False)
          """
          from multiprocessing import Process,Event
          import time , random
          # 1
          '''
          e = Event()
          # 默认属性值是False.
          print(e.is_set()) 
          # 判断内部成员属性是否是False 
          e.wait()
          # 如果是False , 代码程序阻塞
          print(" 代码执行中 ...  ")
          '''
          # 2
          '''
          e = Event()
          # 将这个属性的值改成True
          e.set(编程客栈)
          # 判断内部成员属性是否是True
          e.wait()
          # 如果是True , 代码程序不阻塞
          print(" 代码执行中 ...  ")
          # 将这个属性的值改成False
          e.clear()
          e.wait()
          print(" 代码执行中 .... 2")
          '''
          # 3
          """
          e = Event()
          # wait(3) 代表最多等待3秒;
          e.wait(3)
          print(" 代码执行中 .... 3")
          """
          # ### 模拟经典红绿灯效果
          # 红绿灯切换
          def traffic_light(e):
          	print("红灯亮")
          	while True:
          		if e.is_set():
          			# 绿灯状态 -> 切红灯
          			time.sleep(1)
          			print("红灯亮")
          			# True => False
          			e.clear()
          		else:
          			# 红灯状态 -> 切绿灯
          			time.sleep(1)
          			print("绿灯亮")
          			# False => True
          			e.set()
          # e = Event()
          # traffic_light(e)
          # 车的状态
          def car(e,i):
          	# 判断是否是红灯,如果是加上wait阻塞
          	if not e.is_set():
          		print("car{} 在等待 ... ".format(i))
          		e.wait()
          	# 否则不是,代表绿灯通行;
          	print("car{} 通行了 ... ".format(i))
          """	
          # 1.全国红绿灯
          if __name__ == "__main__":
          	e = Event()
          	# 创建交通灯
          	p1 = Process(target=traffic_light , args=(e,))
          	p1.start()
          	# 创建小车进程
          	for i in range(1,21):
          		time.sleep(random.randrange(2))
          		p2 = Process(target=car , args=(e,i))
          		p2.start()
          """
          # 2.包头红绿灯,没有车的时候,把红绿灯关了,省电;
          if __name__ == "__main__":
          	lst = []
          	e = Event()
          	# 创建交通灯
          	p1 = Process(target=traffic_light , args=(e,))
          	# 设置红绿灯为守护进程
          	p1.daemon = True
          	p1.start()
          	# 创建小车进程
          	for i in range(1,21):
          		time.sleep(random.randrange(2))
          		p2 = Process(target=car , args=(e,i))
          		lst.append(p2)
          		p2.start()
          	# 让所有的小车全部跑完,把红绿灯炸飞
          	print(lst)
          	for i in lst:
          		i.join()
          	print("关闭成功 .... ")
          

          事件知识点:

          # 阻塞事件 :
          	e = Event()生成事件对象e   
          	e.wait()动态给程序加阻塞 , 程序当中是否加阻塞完全取决于该对象中的is_set() [默认返回值是False]
              # 如果是True  不加阻塞
              # 如果是False 加阻塞
          # 控制这个属性的值
              # set()方法     将这个属性的值改成True
              # clear()方法   将这个属性的值改成False
              # is_set()方法  判断当前的属性是否为True  (默认上来是False)
          

          3. queue进程队列

          # ### 进程队列(进程与子进程是相互隔离的,如果两者想要进行通信,可以利用队列实现)
          from multiprocessing import Process,Queue
          # 引入线程模块; 为了捕捉queue.Empty异常;
          import queue
          # 1.基本语法
          """顺序: 先进先出,后进后出"""
          # 创建进程队列
          q = Queue()
          # put() 存放
          q.put(1)
          q.put(2)
          q.put(3)
          # get() 获取
          """在获取不到任何数据时,会出现阻塞"""
          # print(  q.get()  )
          # print(  q.get()  )
          # print(  q.get()  )
          # print(  q.get()  )
          # get_nowait() 拿不到数据报异常
          """[windows]效果正常  [linux]不兼容"""
          try:
          	print(  q.get_nowait()  )
          	print(  q.get_nowait()  )
          	print(  q.get_nowait()  )
          	print(  q.get_nowait()  )
          except : #queue.Empty
          	pass
          # put_nowait() 非阻塞版本的put
          # 设置当前队列最大长度为3 ( 元素个数最多是3个 )
          """在指定队列长度的情况下,如果塞入过多的数据,会导致阻塞"""
          # q2 = Queue(3)
          # q2.put(111)
          # q2.put(222)
          # q2.put(333)
          # q2.put(444)
          """使用put_nowait 在队列已满的情况下,塞入数据会直接报错"""
          q2 = Queue(3)
          try:
          	q2.put_nowait(111)
          	q2.put_nowait(222)
          	q2.put_nowait(333)
          	q2.put_nowait(444)
          except:
          	pass
          # 2.进程间的通信IPC
          def func(q):
          	# 2.子进程获取主进程存放的数据
          	res = q.get()
          	print(res,"<22>")
          	# 3.子进程中存放数据
          	q.put("刘一缝")
          if __name__ == "__main__":
          	q3 = Queue()
          	p = Process(target=func,args=(q3,))
          	p.start()
          	# 1.主进程存入数据
          	q3.put("赵凤勇")
          	# 为了等待子进程把数据存放队列后,主进程在获取数据;
          	p.join()
          	# 4.主进程获取子进程存放的数据
          	print(q3.get() , "<33>")
          

          小提示: 一般主进程比子进程执行的快一些

          Python全栈之队列详解

          队列知识点:

          # 进程间通信 IPC
          # IPC Inter-Process Communication
          # 实现进程之间通信的两种机制:
              # 管道 Pipe
              # 队列 Queue
          # put() 存放
          # get() 获取
          # get_nowait() 拿不到报异常
          # put_nowait() 非阻塞版本的put
          q.empty()      检测是否为空  (了解)
          q.full() 	   检测是否已经存满 (了解)
          

          4. 生产者消费者模型

          # ### 生产者和消费者模型 
          """
          # 爬虫案例
          1号进程负责抓取其他多个网站中相关的关键字信息,正则匹配到队列中存储(mysql)
          2号进程负责把队列中的内容拿取出来,将经过修饰后的内容布局到自个的网站中
          1号进程可以理解成生产者
          2号进程可以理解成消费者
          从程序上来看 
          	生产者负责存储数据 (put)
          	消费者负责获取数据 (get)
          生产者和消费者比较理想的模型:
          	生产多少,消费多少 . 生产数据的速度 和 消费数据的速度 相对一致	
          """
          # 1.基础版生产着消费者模型
          """问题 : 当前模型,程序不能正常终止 """
          """
          from multiprocessing import Process,Queue
          import time,random
          # 消费者模型
          def consumer(q,name):
          	while True:
          		# 获取队列中的数据
          		food = q.get()
          		time.sleep(random.uniform(0.1,1))
          		print("{}吃了{}".format(name,food))
          # 生产者模型
          def producer(q,name,food):
          	for i in range(5):
          		time.sleep(random.uniform(0.1,1))
          		# 展示生产的数据
          		print(  "{}生产了{}".format(  name , food+str(i)  )   )
          		# 存储生产的数据在队列中
          		q.put(food+str(i))
          if __name__ == "__main__":
          	q = Queue()
          	p1 = Process(  target=consumer,args=(q , "赵万里")  )
          	p2 = Process(  target=producer,args=(q , "赵沈阳" , "香蕉" )  )
          	p1.start()
          	p2.start()
          	p2.join()
          """
          # 2.优化模型
          """特点 : 手动在队列的最后,加入标识None, 终止消费者模型"""
          """
          from multiprocessing import Process,Queue
          import time,random
          # 消费者模型
          def consumer(q,name):
          	while True:
          		# 获取队列中的数据
          		food = q.get()
          		# 如果最后一次获取的数据是None , 代表队列已经没有更多数据可以获取了,终止循环;
          		if food is None:
          			break
          		time.sleep(random.uniform(0.1,1))
          		print("{}吃了{}".format(name,food))
          # 生产者模型
          def producer(q,name,food):
          	for i in range(5):
          		time.sleep(random.uniform(0.1,1))
          		# 展示生产的数据
          		print(  "{}生产了{}".format(  name , food+str(i)  )   )
          		# 存储生产的数据在队列中
          		q.put(food+str(i))
          if __name__ == "__main__":
          	q = Queue()
          	p1 = Process(  target=consumer,args编程客栈=(q , "赵万里")  )
          	p2 = Process(  target=producer,args=(q , "赵沈阳" , "香蕉" )  )
          	p1.start()
          	p2.start()
          	p2.join()
          	q.put(None) # 香蕉0 香蕉1 香蕉2 香蕉3 香蕉4 None
          """
          # 3.多个生产者和消费者
          """ 问题 : 虽然可以解决问题 , 但是需要加入多个None  , 代码冗余"""
          from multiprocessing import Process,Queue
          import time,random
          # 消费者模型
          def consumer(q,name):
          	while True:
          		# 获取队列中的数据
          		food = q.get()
          		# 如果最后一次获取的数据是None , 代表队列已经没有更多数据可以获取了,终止循环;
          		if food is None:
          			break
          		time.sleep(random.uniform(0.1,1))
          		print("{}吃了{}".format(name,food))
          # 生产者模型
          def producer(q,name,food):
          	for i in range(5):
          		time.sleep(random.uniform(0.1,1))
          		# 展示生产的数据
          		print(  "{}生产了{}".format(  name , food+str(i)  )   )
          		# 存储生产的数据在队列中
          		q.put(food+str(i))
          if __name__ == "__main__":
          	q = Queue()
          	p1 = Process(  target=consumer,args=(q , "赵万里")  )
          	p1_1 = Process(  target=consumer,args=(q , "赵世超")  )
          	p2 = Process(  target=producer,args=(q , "赵沈阳" , "香蕉" )  )
          	p2_2 = Process(  target=producer,args=(q , "赵凤勇" , "大蒜" )  )
          	p1.start()
          	p1_1.start()
          	p2.start()
          	p2_2.start()
          	# 等待所有数据填充完毕
          	p2.join()
          	p2_2.join()
          	# 把None 关键字放在整个队列的最后,作为跳出消费者循环的标识符;
          	q.put(None) # 给第一个消费者加一个None , 用来终止
          	q.put(None) # 给第二个消费者加一个None , 用来终止
          	# ... 
          

          Python全栈之队列详解

          5. joinablequeue队列使用

          # ### JoinableQueue 队列
          """
          put 存放  
          get 获取  
          task_done 计算器属性值-1  
          join 配合task_done来使用 , 阻塞
          put 一次数据, 队列的内置计数器属性值+1
          get 一次数据, 通过task_done让队列的内置计数器属性值-1
          join: 会根据队列计数器的属性值来判断是否阻塞或者放行
          	队列计数器属性是 等于 0 ,  代码不阻塞放行
          	队列计数器属性是 不等 0 ,  意味着代码阻塞
          """
          from multiprocessing  import JoinableQueue
          jq = JoinableQueue()
          jq.put("王同培") # +1
          jq.put("王伟")   # +2
          print(jq.get())
          print(jq.get())
          # print(jq.get()) 阻塞
          jq.task_done()   # -1
          jq.task_done()   # -1
          jq.join()
          print(" 代码执行结束 .... ")
          # ### 2.使用JoinableQueue 改造生产着消费者模型
          from multiprocessing import Process,Queue
          import time,random
          # 消费者模型
          def consumer(q,name):
          	while True:
          		# 获取队列中的数据
          		food = q.get()
          		time.sleep(random.uniform(0.1,1))
          		print("{}吃了{}".format(name,food))
          		# 让队列的内置计数器属性-1
          		q.task_done()
          #www.cppcns.com 生产者模型
          def producer(q,name,food):
          	for i in range(5):
          		time.sleep(random.uniform(0.1,1))
          		# 展示生产的数据
          		print(  "{}生产了{}".format(  name , food+str(i)  )   )
          		# 存储生产的数据在队列中
          		q.put(food+str(i))
          if __name__ == "__main__":
          	q = JoinableQueue()
          	p1 = Process(  target=consumer,args=(q , "赵万里")  )
          	p2 = Process(  target=producer,args=(q , "赵沈阳" , "香蕉" )  )
          	p1.daemon = True
          	p1.start()
          	p2.start()
          	p2.join()
          	# 必须等待队列中的所有数据全部消费完毕,再放行
          	q.join()
          	print("程序结束 ... ")
          

          6. 总结

          ipc可以让进程之间进行通信
          lock其实也让进程之间进行通信了,多个进程去抢一mvrxQ把锁,一个进程抢到
          这 把锁了,其他的进程就抢不到这把锁了,进程通过socket底层互相发
          消息,告诉其他进程当前状态已经被锁定了,不能再强了。
          进程之间默认是隔离的,不能通信的,如果想要通信,必须通过ipc的
          方式(lock、joinablequeue、Manager)
          

          本篇文章就到这里了,希望能够给你带来帮助,也希望您能够多多关注我们的更多内容!

          0

          精彩评论

          暂无评论...
          验证码 换一张
          取 消

          关注公众号