Python并发编程系列之多进程（multiprocessing）

from multiprocessing import Processimport timedef func (n): print("子进程开始运行:{}……".format(n)) time.sleep(1) print("子进程结束运行:{}……".format(n))if __name__ == '__main__':#创建进程执行一定要这一行代码 print("主进程开始运行……") p = Process(target=func,args=(10,)) # 注册 p.start()# 启动一个子进程 time.sleep(1)print("主进程结束运行……" )

from multiprocessing import Processimport osclass MyProcess(Process): def run(self):#必须覆写run方法 print('子进程:' , self.pid)if __name__=="__main__": p1 = MyProcess() p1.start() p2 = MyProcess() p2.start()print('主进程：', os.getpid())

from multiprocessing import Processimport osclass MyProcess(Process): def __init__(self , arg1 , arg2): super().__init__() self.arg1 = arg1 self.arg2 = arg2 def run(self): print('子进程:' , self.pid , self.arg1 , self.arg2) if __name__=="__main__": p1 = MyProcess('我是arg1' , '我是arg2') p1.start() p2 = MyProcess('我是arg1' , '我是arg2') p2.start()print('主进程：', os.getpid())

import timeimport osfrom multiprocessing import Processdef func(): i = 1 while True: time.sleep(1) print('{}--子进程p1正在执行,pid:{}'.format(i , os.getpid())) i+=1def func2(): print('子进程p2开始执行,pid:{}'.format(os.getpid())) time.sleep(10) print('子进程p2结束执行,pid:{}'.format(os.getpid()))if __name__=='__main__': print('主进程代码开始运行,pid:{}'.format(os.getpid())) p = Process(target=func) p.daemon = True # 设置为守护进程 p.start() p2 = Process(target=func2) p2.start() time.sleep(5)print('主进程代码运行完了,pid:{}'.format(os.getpid()))

import timeimport osfrom multiprocessing import Processdef func(): i = 1 while True: time.sleep(1) print('{}--子进程p1正在执行,pid:{}'.format(i , os.getpid())) i+=1if __name__=='__main__': p = Process(target=func) p.start() time.sleep(3) p.terminate() # 终结进程p print(p.is_alive()) # 检查p是否依然存活 time.sleep(1) print(p.is_alive())

from multiprocessing import Processimport timedef func1 (): print("进程1开始运行……") for i in range(3): time.sleep(1) print("进程1运行了{}秒……".format(i+1)) print("进程1结束运行……")def func2 (): print("进程2开始运行……") for i in range(6): time.sleep(1) print("进程2运行了{}秒……".format(i+1)) print("进程2结束运行……")if __name__ == '__main__': print("主进程开始运行……") p1 = Process(target=func1) p2 = Process(target=func2) p1.start() p2.start() time.sleep(1) p1.join() # p2.join() print("主进程结束运行……" )

from multiprocessing import Processfrom multiprocessing import Lockimport timeimport osdef func (): if os.path.exists('num.txt'): with open('num.txt' , 'r') as f: num = int(f.read()) num -= 1 time.sleep(1) with open('num.txt' , 'w') as f: f.write(str(num)) else: with open('num.txt' , 'w') as f: f.write('')if __name__ == '__main__': print("主进程开始运行……") p_list = [] for i in range(10): p = Process(target=func) p_list.append(p) p.start() for p in p_list: p.join() with open('num.txt', 'r') as f: num = int(f.read()) print('最后结果为：{}'.format(num)) print("主进程结束运行……" )

from multiprocessing import Processfrom multiprocessing import Lockimport timeimport osdef func (lock): if os.path.exists('num.txt'): lock.acquire() with open('num.txt' , 'r') as f: num = int(f.read()) num -= 1 time.sleep(1) with open('num.txt' , 'w') as f: f.write(str(num)) lock.release() else: with open('num.txt' , 'w') as f: f.write('')if __name__ == '__main__': print("主进程开始运行……") lock = Lock() p_list = [] for i in range(10): p = Process(target=func , args=(lock,)) p_list.append(p) p.start() for p in p_list: p.join() with open('num.txt', 'r') as f: num = int(f.read()) print('最后结果为：{}'.format(num)) print("主进程结束运行……" )

from multiprocessing import Processimport timeimport randomdef fun(i): print('{}号顾客上座，开始吃饭'.format(i)) time.sleep(random.randint(3, 8)) print('{}号顾客吃完饭了，离座'.format(i))if __name__=='__main__': for i in range(100): p = Process(target=fun, args=(i,)) p.start()

from multiprocessing import Processimport timeimport randomfrom multiprocessing import Semaphoredef fun(i , sem): sem.acquire() print('{}号顾客上座，开始吃饭'.format(i)) time.sleep(random.randint(3, 8)) print('{}号顾客吃完饭了，离座'.format(i)) sem.release()if __name__=='__main__': sem = Semaphore(3) for i in range(10): p = Process(target=fun, args=(i,sem)) p.start()

from multiprocessing import Event, Processimport timefrom datetime import datetimedef func(e): print('子进程：开始运行……') while True: print('子进程：现在事件秒数是{}'.format(datetime.now().second)) e.wait() # 阻塞等待信号 这里插入了一个flag 默认为 False time.sleep(1)if __name__ == '__main__': e = Event() p = Process(target=func, args=(e,)) p.start() for i in range(10): s = int(str(datetime.now().second)[-1])#获取当前秒数的个位数 if s < 5: print('子进程进入阻塞状态') e.clear() # 使插入的flag为False 进程进入阻塞状态 else: print('子进程取消阻塞状态') e.set() # 使插入的flag为True，进程进入非阻塞状态 time.sleep(1) e.set() time.sleep(3) p.terminate() print("主进程运行结束……")

from multiprocessing import Processfrom multiprocessing import Queueimport randomimport os# 向queue中输入数据的函数def inputQ(queue ): info = random.randint(1,100) queue.put(info) print('进程{}往队列中存了一个数据：{}'.format(os.getpid() , info))# 向queue中输出数据的函数def outputQ(queue): info = queue.get() print ('进程{}从队列中取出一个数据：{}'.format(os.getpid() , info))if __name__ == '__main__': queue = Queue(5) lst_1 = [] lst_2 = [] for i in range(3): process = Process(target=inputQ,args=(queue,)) process.start() lst_1.append(process) # 输出进程 for i in range(2): process = Process(target=outputQ ,args=(queue,)) process.start() lst_1.append(process) for p in lst_2: p.join() for p in lst_2: p.join()

from multiprocessing import Process, Pipedef f(conn): conn.send('主进程，你好呀！') # 发送数据给主进程 print('子进程收到主进程发来的数据：{}'.format(conn.recv())) conn.close() # 关闭if __name__ == '__main__': parent_conn, child_conn = Pipe() #Pipe是一个函数，返回的是一个元组 p = Process(target=f, args=(child_conn,)) # 创建一个子进程 p.start() print("主进程收到子进程发来的数据：{}".format(parent_conn.recv())) parent_conn.send('子进程，你好啊！') # 发送数据给子进程p.join()

from multiprocessing import Processimport osdef func(): global n n = 10 print('子进程pid:{},n:{}'.format(os.getppid() , n))if __name__=='__main__': n=100 print('主进程pid:{},n:{}'.format(os.getppid(), n)) p = Process(target=func) p.start() p.join() print('主进程中输出n：{}'.format(n))

import osfrom multiprocessing import Manager,Process# 定义了一个foo函数,接收一个字典和一个列表def foo(d, l): # 字典和列表都放进程ID d[os.getpid()] = os.getpid() l.append(os.getpid())if __name__ == '__main__': # 生成Manager对象 manager = Manager() d = manager.dict() l = manager.list(range(3)) # 10个进程分别join p_list = [] for i in range(10): p = Process(target=foo, args=(d, l)) p.start() p_list.append(p) for res in p_list: res.join() # 打印字典和列表 print(d) print(l)

import os,time,randomfrom multiprocessing import Pooldef func1(n): print('任务{}开始执行，进程为：{}'.format(n,os.getpid())) time.sleep(random.randint(1,4)) print('任务{}结束执行，进程为：{}'.format(n,os.getpid()))if __name__ == '__main__': p=Pool(3) #c创建一个进程池，里面有三个进程 for i in range(10): res=p.apply(func1,args=(i,))

import os,time,random
from multiprocessing import Pooldef func1(n):
print('任务{}开始执行，进程为：{}'.format(n,os.getpid())) time.sleep(random.randint(1,4)) print('任务{}结束执行，进程为：{}'.format(n,os.getpid()))if __name__ == '__main__': p=Pool(3) #c创建一个进程池，里面有三个进程 for i in range(5): res=p.apply_async(func1,args=(i,)) p.close()#一定要关闭 p.join()#一定要使用join，不然进程池里的进程没来得及执行，主进程结束了，子进程也都跟着结束。

from multiprocessing import Pooldef func1(n): return n**2if __name__ == '__main__': p=Pool(3) result_lst = [] for i in range(5): result_lst.append(p.apply_async(func1,args=(i,)))#异步 p.close() p.join() for result in result_lst: print(result.get())#如果是同步，就不用get了，直接用result获取