#创建一个子进程,并将参数传入 import os import time from multiprocessing import Process def func(args,args2): print(args,args2) time.sleep(3) print("子进程",os.getpid()) print("子进程的父进程", os.getppid()) print(123456) if __name__ == "__main_": p = Process(target=func,args=("参数","参数2"))#注册 #p是一个进程对象,还没有启动进程 p.start()#开启了一个子进程 print("*"*10) print("父进程",os.getpid()) print("父进程的父进程",os.getppid())
进程的生命周期 # 主进程 # 子进程 # 开启了子进程的主进程 : # 主进程自己的代码如果长,等待自己的代码执行结束, # 子进程的执行时间长,主进程会在主进程代码执行完毕之后等待子进程执行完毕之后 主进程才结束
#面向对象创 import os from multiprocessing import Process class MyProcess(Process): def __init__(self,arg1,arg2): super().__init__() self.arg1 = arg1 self.arg2 = arg2 def run(self): print(self.pid) print(self.name) print(self.arg1) print(self.arg2) if __name__ == "__main__": p1 = MyProcess(1,1) p1.start() p2 = MyProcess(1,2) p2.start()
import os import time from multiprocessing import Process def func(filename,content): with open(filename,"w")as f: f.write(content*10*"*") if __name__=="__main__": p_lst = [] for i in range(10): p = Process(target=func,args=("info%s"%i,0)) p_lst.append(p) p.start() for p in p_lst:p.join() print([i for i in os.walk(r'E:python10day37')])
进程与进程之间不经过操作的话,相互隔离数据
# 子进程 -- > 守护进程 import time from multiprocessing import Process def func(): while True: time.sleep(0.2) print('我还活着') def func2(): print('in func2 start') time.sleep(8) print('in func2 finished') if __name__ == '__main__': p = Process(target=func) p.daemon = True # 设置子进程为守护进程 p.start() p2 = Process(target=func2) p2.start() p2.terminate() # 结束一个子进程 time.sleep(1) print(p2.is_alive()) # 检验一个进程是否还活着 print(p2.name) # i = 0 # while i<5: # print('我是socket server') # time.sleep(1) # i+=1 # 守护进程 会 随着 主进程的代码执行完毕 而 结束 # 在主进程内结束一个子进程 p.terminate() # 结束一个进程不是在执行方法之后立即生效,需要一个操作系统响应的过程 # 检验一个进程是否活着的状态 p.is_alive() # p.name p.pid 这个进程的名字和进程号
import time from multiprocessing import Process from multiprocessing import Lock # def show(i): # with open('ticket') as f: # dic = json.load(f) # print('余票: %s'%dic['ticket']) def buy_ticket(i,lock): lock.acquire() #拿钥匙进门 with open('ticket') as f: dic = json.load(f) time.sleep(0.1) if dic['ticket'] > 0 : dic['ticket'] -= 1 print(' 33[32m%s买到票了 33[0m'%i) else: print(' 33[31m%s没买到票 33[0m'%i) time.sleep(0.1) with open('ticket','w') as f: json.dump(dic,f) lock.release() # 还钥匙 if __name__ == '__main__': # for i in range(10): # p = Process(target=show,args=(i,)) # p.start() lock = Lock() for i in range(10): p = Process(target=buy_ticket, args=(i,lock)) p.start()
多进程锁的信号量
#多进程实现一个代码让特定数量的用户进行执行的方法 # 用到模块multiprocessing模块 # Semaphore(设置能够进入函数的用户数量) import time,random from multiprocessing import Process,Semaphore sem = Semaphore(2) sem.acquire()#第一个 sem.acquire()#第二个 def ktv(i,sem): sem.acquire()#拿钥匙 sem.release()#还钥匙 if __name__ == '__main__' : sem = Semaphore()#创建一个类 for i in range(20): p=Process(target=ktv,args=(i,sem))#创建一个进程 p.start()#开启一个进程
事件基础知识
# 事件 # 通过一个信号来控制多个进程同时执行或者阻塞 from multiprocessing import Event # 用到multiprocess模块 # 用到Event方法 e = Event()#创建一个事件 e.is_set()#查看事件状态,默认是阻塞状态False e.set()#将事件状态改为True e.clear()#将事件变为阻塞状态false e.wait()#根据e.is_set()的值来决定是否阻塞 False阻塞 True不阻塞
# 红绿灯事件 import time import random from multiprocessing import Event,Process def cars(e,i): if not e.is_set(): print('car%i在等待'%i) e.wait() # 阻塞 直到得到一个 事件状态变成 True 的信号 print(' 33[0;32;40mcar%i通过 33[0m' % i) def light(e): while True: if e.is_set(): e.clear() print(' 33[31m红灯亮了 33[0m') else: e.set() print(' 33[32m绿灯亮了 33[0m') time.sleep(2) if __name__ == '__main__': e = Event() traffic = Process(target=light,args=(e,)) traffic.start() for i in range(20): car = Process(target=cars, args=(e,i)) car.start() time.sleep(random.random())
队列基础知识
#队列 from multiprocessing import Queue # 用到multiprocessing模块 # 用到Queue方法 q = Queue()#创建队列 q.put()#加入队列 q.get()#退出队列 q.empty()#用于判断队列是否空了 q.full()#用于判断队列是否满了
# 队列 先进先出 # IPC # from multiprocessing import Queue # q = Queue(5) # q.put(1) # q.put(2) # q.put(3) # q.put(4) # q.put(5) # print(q.full()) # 队列是否满了 # print(q.get()) # print(q.get()) # print(q.get()) # print(q.get()) # print(q.get()) # print(q.empty()) # while True: # try: # q.get_nowait() # except: # print('队列已空') # time.sleep(0.5) # for i in range(6): # q.put(i) from multiprocessing import Queue,Process def produce(q): q.put('hello') def consume(q): print(q.get()) if __name__ == '__main__': q = Queue() p = Process(target=produce,args=(q,)) p.start() c = Process(target=consume, args=(q,)) c.start()
# 队列 # 生产者消费者模型 # 生产者 进程 # 消费者 进程 import time import random from multiprocessing import Process,Queue def consumer(q,name): while True: food = q.get() if food is None: print('%s获取到了一个空'%name) break print(' 33[31m%s消费了%s 33[0m' % (name,food)) time.sleep(random.randint(1,3)) def producer(name,food,q): for i in range(4): time.sleep(random.randint(1,3)) f = '%s生产了%s%s'%(name,food,i) print(f) q.put(f) if __name__ == '__main__': q = Queue(20) p1 = Process(target=producer,args=('Egon','包子',q)) p2 = Process(target=producer, args=('wusir','泔水', q)) c1 = Process(target=consumer, args=(q,'alex')) c2 = Process(target=consumer, args=(q,'jinboss')) p1.start() p2.start() c1.start() c2.start() p1.join() p2.join() q.put(None) q.put(None)
import time import random from multiprocessing import Process,JoinableQueue def consumer(q,name): while True: food = q.get() print(' 33[31m%s消费了%s 33[0m' % (name,food)) time.sleep(random.randint(1,3)) q.task_done() # count - 1 def producer(name,food,q): for i in range(4): time.sleep(random.randint(1,3)) f = '%s生产了%s%s'%(name,food,i) print(f) q.put(f) q.join() # 阻塞 直到一个队列中的所有数据 全部被处理完毕 if __name__ == '__main__': q = JoinableQueue(20) p1 = Process(target=producer,args=('Egon','包子',q)) p2 = Process(target=producer, args=('wusir','泔水', q)) c1 = Process(target=consumer, args=(q,'alex')) c2 = Process(target=consumer, args=(q,'jinboss')) p1.start() p2.start() c1.daemon = True # 设置为守护进程 主进程中的代码执行完毕之后,子进程自动结束 c2.daemon = True c1.start() c2.start() p1.join() p2.join() # 感知一个进程的结束 # 在消费者这一端: # 每次获取一个数据 # 处理一个数据 # 发送一个记号 : 标志一个数据被处理成功 # 在生产者这一端: # 每一次生产一个数据, # 且每一次生产的数据都放在队列中 # 在队列中刻上一个记号 # 当生产者全部生产完毕之后, # join信号 : 已经停止生产数据了 # 且要等待之前被刻上的记号都被消费完 # 当数据都被处理完时,join阻塞结束 # consumer 中把所有的任务消耗完 # producer 端 的 join感知到,停止阻塞 # 所有的producer进程结束 # 主进程中的p.join结束 # 主进程中代码结束 # 守护进程(消费者的进程)结束