fork方式創(chuàng)建進(jìn)程
簡(jiǎn)單的fork
import os
ret = os.fork()
print("haha")
主進(jìn)程fork時(shí)返回值大于0删性，子進(jìn)程fork時(shí)返回值等于0
import os
import time
ret = os.fork()
if ret == 0:
while True:
print("----1----")
time.sleep(1)
else:
while True:
print("----2----")
time.sleep(1)
os.getpid()獲取本進(jìn)程pid蝙搔，os.getppid()獲取本進(jìn)程的父pid翘鸭，os.fork()返回值是生成的子進(jìn)程的pid
import os
ret = os.fork()
print("haha")
if ret > 0:
print("---父進(jìn)程---%d" % os.getpid())
else:
print("---子進(jìn)程---%d---%d" % (os.getpid(), os.getppid()))
print(ret)
主進(jìn)程結(jié)束不會(huì)等子進(jìn)程
import os
import time
"""
begin會(huì)執(zhí)行一次
over會(huì)執(zhí)行兩次
"""
print("---begin---")
ret = os.fork()
if ret == 0:
print("---子進(jìn)程---")
time.sleep(1)
print("---子進(jìn)程over---")
else:
print("---父進(jìn)程---")
time.sleep(3)
print("---主進(jìn)程over---")
print("---over----")
全局變量在多個(gè)進(jìn)程中不共享
import os
import time
g_num = 100
ret = os.fork()
if ret == 0:
print("---process-1---")
g_num += 1
print("---process-1 g_num=%d---" % g_num)
else:
time.sleep(3)
print("---process-2---")
print("---process-2 g_num=%d---" % g_num)
輸出：
---process-1---
---process-1 g_num=101---
---process-2---
---process-2 g_num=100---
多次fork
"""
2*2*2 = 8
8個(gè)進(jìn)程
"""
import os
os.fork()
os.fork()
os.fork()
print("1111")
Process方式創(chuàng)建進(jìn)程
Process([target,[group[,name[,args[,kwargs]]]]])
- target：調(diào)用的對(duì)象
- args：調(diào)用對(duì)象的參數(shù)元組
- kwargs：調(diào)用對(duì)象的參數(shù)字典
- name：當(dāng)前進(jìn)程實(shí)例的別名
- group
類方法：
- is_alive() 判斷進(jìn)程實(shí)例是否還在執(zhí)行
- join([timeout]) 是否等待子進(jìn)程執(zhí)行結(jié)束或等待多少秒
- start() 啟動(dòng)進(jìn)程實(shí)例，即開始創(chuàng)建子進(jìn)程
- run() 如果沒有給定target參數(shù)，對(duì)這個(gè)對(duì)象調(diào)用start()方法時(shí)瓷胧，就將執(zhí)行對(duì)象中的run()方法
- terminate() 不管任務(wù)是否完成，立即終止
Process類常用屬性：
- name：當(dāng)前進(jìn)程實(shí)例別名棚愤，默認(rèn)為Process-N搓萧，N為從1開始遞增的整數(shù)
- pid：當(dāng)前進(jìn)程實(shí)例的PID值
from multiprocessing import Process
import time
def test():
while True:
print("---test---")
time.sleep(1)
p = Process(target=test)
# 讓這個(gè)進(jìn)程開始執(zhí)行test函數(shù)里的代碼
p.start()
while True:
print("---main---")
time.sleep(1)
一般情況下主進(jìn)程不會(huì)等子進(jìn)程結(jié)束再結(jié)束，而是直結(jié)束
from multiprocessing import Process
import time
def test():
for i in range(5):
print("---test---")
time.sleep(1)
p = Process(target=test)
p.start() # 讓這個(gè)進(jìn)程開始執(zhí)行test函數(shù)里的代碼
print("結(jié)束")
輸出：
結(jié)束
---test---
---test---
---test---
---test---
---test---
使用join方法宛畦，可以讓主進(jìn)程堵塞瘸洛，等待所有子進(jìn)程結(jié)束后，主進(jìn)程結(jié)束
from multiprocessing import Process
import time
import random
def test():
for i in range(random.randint(1,5)):
print("---%d---" % i)
time.sleep(1)
p = Process(target = test)
p.start()
p.join()
print("---main---")
輸出：
---0---
---1---
---2---
---3---
---4---
---main---
通過(guò)Process子類的方式創(chuàng)建進(jìn)程
from multiprocessing import Process
import time
class MyNewProcess(Process):
def run(self):
while True:
print("---1---")
time.sleep(1)
p = MyNewProcess()
p.start()
while True:
print("---main---")
time.sleep(1)
進(jìn)程池
當(dāng)需要?jiǎng)?chuàng)建的進(jìn)程數(shù)量不多時(shí)次和，可以直接使用multiprocessing中的Process創(chuàng)建多個(gè)進(jìn)程反肋，如果是成百上千的目標(biāo)，可以利用multiprocessing中Pool進(jìn)程池的方法提高效率
multiprocessing.Pool常用方法：
- apply_async(func[,args[,kwds]])：使用非阻塞方式調(diào)用func
- apply(func[,args[,kwds]])：使用阻塞方式調(diào)用func
- terminate()：不管任務(wù)是否完成踏施，立即終止
- join()：主進(jìn)程阻塞石蔗，等待子進(jìn)程的退出，必須在close或terminate之后使用
apply_async方式
from multiprocessing import Pool
import os
import time
def worker(n):
for i in range(n):
time.sleep(1)
print("---pid = %d---n = %d" % (os.getpid(), i))
pool = Pool(5)
for i in range(10):
print("---%d---" % i)
pool.apply_async(worker,(5,))
print("---start---")
pool.close() # 關(guān)閉進(jìn)程池畅形，相當(dāng)于不能夠再添加新任務(wù)了
pool.join()
print("---end---")
apply方式
from multiprocessing import Pool
import os
import time
def worker(n):
for i in range(n):
time.sleep(1)
print("---pid = %d---n = %d" % (os.getpid(), i))
pool = Pool(5)
for i in range(10):
print("---%d---" % i)
# 堵塞的方式
pool.apply(worker,(5,))
print("---start---")
pool.close() # 關(guān)閉進(jìn)程池养距，相當(dāng)于 不能夠再添加新任務(wù)了
pool.join()
print("---end---")
進(jìn)程間通信-Queue
Queue本身是一個(gè)消息隊(duì)列程序，可以利用它進(jìn)行進(jìn)程間通信
- Queue.size() 返回當(dāng)前隊(duì)列包含的消息數(shù)量（mac用不了）
- Queue.empty() 隊(duì)列為空返回True日熬，反之返回False
- Queue.full() 隊(duì)列滿了返回True棍厌，反之返回False
- Queue.get([block[,timeout]]) 獲取隊(duì)列中一條消息，然后將其從隊(duì)列中移除竖席，block默認(rèn)為True
- block默認(rèn)為True耘纱，如果消息隊(duì)列為空，且沒有設(shè)置timeout毕荐，則程序被阻塞束析，直到消息隊(duì)列讀到消息為止
- block為False時(shí)，如果消息隊(duì)列為空憎亚，立即拋出Queue.Empty異常
- timeout员寇，超時(shí)時(shí)間，沒有讀到消息時(shí)會(huì)等待timeout秒虽填，如果還沒讀到丁恭，就拋出Queue.Empty異常
- Queue.get_nowait() 相當(dāng)于Queue.get(False)
- Queue.put(item[,block[,timeout]]) 將消息寫入隊(duì)列，block默認(rèn)為True
- block默認(rèn)為True斋日，如果消息隊(duì)列沒有寫入空間牲览，且沒有設(shè)置timeout，則程序被阻塞，直到消息隊(duì)列能寫入為止
- block為False時(shí)第献，如果消息隊(duì)列沒有寫入空間贡必，立即拋出Queue.Full異常
- timeout，超時(shí)時(shí)間庸毫，沒有寫入空間時(shí)時(shí)會(huì)等待timeout秒仔拟，如果還沒有寫入空間，就拋出Queue.Full異常
- Queue.put_nowait() 相當(dāng)于Queue.put(item, False)
Process 下的Queue通信
from multiprocessing import Queue, Process
import os, time
# 寫數(shù)據(jù)
def write(q):
for value in ["A", "B", "C"]:
q.put(value)
print("put queue %s" % value)
time.sleep(1)
# 讀數(shù)據(jù)
def read(q):
while True:
if not q.empty():
value = q.get()
print("get queue %s " % value)
time.sleep(1)
else:
break
if __name__ == "__main__":
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
pw.start()
pw.join()
pr.start()
pr.join()
print("結(jié)束")
Pool下的Queue通信
進(jìn)程池中的Queue需要使用multiprocess.Manager()中的Queue飒赃，而不是multiprocess.Queue()
from multiprocessing import Manager, Pool
import os, time
def write(q):
for value in ["A", "B", "C"]:
q.put(value)
time.sleep(1)
print("queue put %s " % value)
def read(q):
while True:
if not q.empty():
value = q.get()
time.sleep(1)
print("queue get %s " % value)
else:
break
if __name__ == "__main__":
q = Manager().Queue()
pool = Pool()
pool.apply_async(write, (q,))
pool.apply_async(read, (q,))
pool.close()
pool.join()
print("結(jié)束")
