守护进程
主进程创建守护进程
其一：守护进程会在主进程代码执行结束后就终止
其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止
# 守护进程
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())
if __name__ == ‘__main__‘:
p = Process(target=task)
p.daemon = True # 必须在p.start()之前，守护进程不能开启子进程
p.start()
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 threading import Thread
import os,time,random
def task():
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 # 必须在p.start()之前
t.start()
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-------")
互斥锁
# 进程
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())
mutex.release()
if __name__ == ‘__main__‘:
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)
加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。 虽然可以用文件共享数据实现进程间通信，但问题是：
1.效率低（共享数据基于文件，而文件是硬盘上的数据）
2.需要自己加锁处理
因此我们最好找寻一种解决方案能够兼顾：1、效率高（多个进程共享一块内存的数据）2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
1 队列和管道都是将数据存放于内存中
2 队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来
我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性
模拟抢票
# 创建db.txt文件
# {"count": 1}
from multiprocessing import Process,Lock
import json,os,time,random
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__‘:
mutex = Lock()
for i in range(20):
p = Process(target=task,args=(mutex,))
p.start()
信号量
互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去，如果指定信号量为3，那么来一个人获得一把锁，计数加1，当计数等于3时，后面的人均需要等待。一旦释放，就有人可以获得一把锁，信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念
from multiprocessing import Process,Semaphore
import time,random
def go_wc(sem,user):
sem.acquire()
print(‘%s 占到一个茅坑‘ %user)
time.sleep(random.randint(0,3)) #模拟每个人拉屎速度不一样，0代表有的人蹲下就起来了
sem.release()
if __name__ == ‘__main__‘:
sem=Semaphore(5)
p_l=[]
for i in range(13):
p=Process(target=go_wc,args=(sem,‘user%s‘ %i,))
p.start()
p_l.append(p)
for i in p_l:
i.join()
print(‘============》‘)
进程队列和线程队列
from multiprocessing import Queue # 进程队列
import queue # 线程队列
q = Queue(3)
q.put({‘a‘:1})
q.put(‘xxxx‘)
q.put(3)
print(q.get())
print(q.get())
print(q.get())
# 队列
q = queue.Queue(3)
q.put({‘a‘:1})
q.put(‘xxxx‘)
q.put(3)
print(q.get())
print(q.get())
print(q.get())
# 优先级队列，数字越小，优先级越高
q = queue.PriorityQueue(3)
q.put((10,{‘a‘:1}))
q.put((-1,‘xxxx‘))
q.put((0,3))
print(q.get())
print(q.get())
print(q.get())
# 堆栈
q = queue.LifoQueue(3)
q.put({‘a‘:1})
q.put(‘xxxx‘)
q.put(3)
print(q.get())
print(q.get())
print(q.get())
生产者消费者模型
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
from multiprocessing import Queue,Process
import time,random,os
def producer(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=producer,args=(q,))
c = Process(target=consumer,args=(q,))
p.start()
c.start()
p.join()
q.put(None)
print(‘主‘)
时间： 2024-10-12 01:17:38