關(guān)于python多進(jìn)程中的常用方法詳解
更新時(shí)間：2023年07月26日 09:03:45 作者：IT之一小佬
這篇文章主要介紹了關(guān)于python多進(jìn)程中的常用方法詳解,python中的多線程其實(shí)并不是真正的多線程，如果想要充分地使用多核CPU資源，在python中大部分情況需要使用多進(jìn)程,需要的朋友可以參考下
1、進(jìn)程間的通信
全局變量在多個(gè)進(jìn)程中不共享資源，進(jìn)程之間的數(shù)據(jù)是獨(dú)立的，默認(rèn)情況下是互不影響的。
示例代碼：
from multiprocessing import Process
num = 1
def task1():
global num
num += 5
print("子進(jìn)程1運(yùn)行，num:", num)
def task2():
global num
num += 10
print("子進(jìn)程2運(yùn)行，num:", num)
if __name__ == '__main__':
print("父進(jìn)程開(kāi)始運(yùn)行...")
p1 = Process(target=task1)
p2 = Process(target=task2)
p1.start()
p2.start()
p1.join()
p2.join()運(yùn)行結(jié)果：
2、用 Queue()實(shí)現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞
Queue 是多進(jìn)程安全的隊(duì)列，可以使用 Queue 實(shí)現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞。
put 方法用以插入數(shù)據(jù)到隊(duì)列中， put 方法還有兩個(gè)可選參數(shù)： blocked 和 timeout。如果 blocked 為 True（默認(rèn)值），并且 timeout 為正值，該方法會(huì)阻塞 timeout 指定的時(shí)間，直到該隊(duì)列有剩余的空間。如果超時(shí)，會(huì)拋出 Queue.full 異常。如果 blocked 為 False，但該 Queue 已滿，會(huì)立即拋出 Queue.full 異常。
get 方法可以從隊(duì)列讀取并且刪除一個(gè)元素。同樣， get 方法有兩個(gè)可選參數(shù)： blocked和 timeout。如果 blocked 為 True（默認(rèn)值），并且 timeout 為正值，那么在等待時(shí)間內(nèi)沒(méi)有取到任何元素，會(huì)拋出 Queue.Empty 異常。如果 blocked 為 False，有兩種情況存在，如果Queue 有一個(gè)值可用，則立即返回該值，否則，如果隊(duì)列為空，則立即拋出Queue.Empty 異常
示例代碼：
from multiprocessing import Queue
q = Queue(3)
q.put('msg1')
q.put('msg2')
print('消息隊(duì)列是否已滿：', q.full())
q.put('msg3')
print('消息隊(duì)列是否已滿：', q.full())
# q.put('msg4') # 以為消息隊(duì)列已經(jīng)滿了，需要直接寫(xiě)入需要等待，如果超時(shí)會(huì)拋出異常
# 寫(xiě)入數(shù)據(jù)時(shí)先判斷，判斷隊(duì)列是否已滿
if not q.full():
q.put('msg4')
# 同理，取消息時(shí)可以先判斷隊(duì)列是否有數(shù)據(jù)
if not q.empty():
for _ in range(q.qsize()):
print(q.get())運(yùn)行結(jié)果：
示例代碼：
from multiprocessing import Process
from multiprocessing import Queue
import random
import os
# 向queue中輸入數(shù)據(jù)的函數(shù)
def inputQ(queue):
info = random.randint(1, 100)
queue.put(info)
print('進(jìn)程{}往隊(duì)列中存了一個(gè)數(shù)據(jù)：{}'.format(os.getpid(), info))
# 向queue中輸出數(shù)據(jù)的函數(shù)
def outputQ(queue):
info = queue.get()
print('進(jìn)程{}從隊(duì)列中取出一個(gè)數(shù)據(jù)：{}'.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)
# 輸出進(jìn)程
for i in range(2):
process = Process(target=outputQ, args=(queue,))
process.start()
lst_2.append(process)
for p in lst_1:
p.join()
for p in lst_2:
p.join()運(yùn)行結(jié)果：
3、Queue 隊(duì)列實(shí)現(xiàn)進(jìn)程間通信
示例代碼：
import time
from multiprocessing import Queue, Process
def write_data(q):
# 將列表元素寫(xiě)入到隊(duì)列中
for i in ['aa', 'bb', 'cc', 'dd']:
print('開(kāi)始寫(xiě)入值%s' %i)
q.put(i)
time.sleep(1)
def read_data(q):
print("開(kāi)始讀取數(shù)據(jù)...")
while True:
if not q.empty():
print("讀取到數(shù)據(jù)：", q.get())
time.sleep(1)
else:
break
if __name__ == '__main__':
# 創(chuàng)建隊(duì)列
q = Queue()
# 創(chuàng)建進(jìn)程
qw = Process(target=write_data, args=(q, ))
qr = Process(target=read_data, args=(q, ))
# 啟動(dòng)進(jìn)程
qw.start()
qr.start()
qw.join()
qr.join()運(yùn)行結(jié)果：
4、Manage()的使用
如果使用 Pool 創(chuàng)建進(jìn)程，就需要使用 multiprocessing.Manager()中的 Queue()來(lái)完成進(jìn)程間的通信，而不是 multiprocessing.Queue()，否則會(huì)拋出異常。
示例代碼：
import time
from multiprocessing import Manager, Pool, Queue
def write_data(q):
# 將列表元素寫(xiě)入到隊(duì)列中
for i in ['aa', 'bb', 'cc', 'dd']:
print('開(kāi)始寫(xiě)入值%s' %i)
q.put(i)
time.sleep(1)
def read_data(q):
print("開(kāi)始讀取數(shù)據(jù)...")
while True:
if not q.empty():
print("讀取到數(shù)據(jù)：", q.get())
time.sleep(1)
else:
break
if __name__ == '__main__':
# 創(chuàng)建隊(duì)列
q = Manager().Queue()
# q = Queue() # 直接這樣使用Queue()會(huì)報(bào)錯(cuò)
# 創(chuàng)建進(jìn)程池
p = Pool(3)
# 使用apply阻塞模式創(chuàng)建進(jìn)程
p.apply(write_data, (q, ))
p.apply(read_data, (q, ))
p.close()
p.join()運(yùn)行結(jié)果：
示例代碼：
import os
from multiprocessing import Manager, Process
# 定義了一個(gè)foo函數(shù),接收一個(gè)字典和一個(gè)列表
def foo(dic, lst):
# 字典和列表都放進(jìn)程ID
dic[os.getpid()] = os.getpid()
lst.append(os.getpid())
if __name__ == '__main__':
# 生成Manager對(duì)象
manager = Manager()
dic = manager.dict()
print(dic)
lst = manager.list(range(3))
print(lst)
# 10個(gè)進(jìn)程分別join
p_list = []
for i in range(10):
p = Process(target=foo, args=(dic, lst))
p.start()
p_list.append(p)
for res in p_list:
res.join()
# 打印字典和列表
print(dic)
print(lst)運(yùn)行結(jié)果：
示例代碼：
from multiprocessing import Pool, current_process, Manager
import time
def produce_data(queue):
for i in range(10):
queue.put(i)
def consume_data(queue):
while queue.qsize() > 0:
data = queue.get() # 注意：當(dāng)get()拿不到數(shù)據(jù)時(shí)，會(huì)一直處于等待狀態(tài)
print(f"當(dāng)前進(jìn)程為：{current_process().name}, 隊(duì)列獲取數(shù)據(jù)為：{data},隊(duì)列剩余數(shù)據(jù)為：{queue.qsize()}個(gè)！")
time.sleep(0.01)
if __name__ == '__main__':
print(f"主進(jìn)程{current_process().name}開(kāi)始執(zhí)行！")
p = Pool(processes=6, maxtasksperchild=6)
queue = Manager().Queue(maxsize=20)
p.apply_async(produce_data, args=(queue, ))
time.sleep(1)
for i in range(5):
p.apply_async(consume_data, args=(queue, ))
p.close() # 關(guān)閉進(jìn)程池，防止將任何其他任務(wù)提交到池中。需要在join之前調(diào)用，否則會(huì)報(bào)ValueError: Pool is still running錯(cuò)誤
p.join() # 等待進(jìn)程池中的所有進(jìn)程執(zhí)行完畢
print(f"主進(jìn)程{current_process().name}結(jié)束！")運(yùn)行結(jié)果：
注意：
- p.close() # 關(guān)閉進(jìn)程池，防止將任何其他任務(wù)提交到池中。需要在join之前調(diào)用，否則會(huì)報(bào)ValueError: Pool is still running錯(cuò)誤
- p.join() # 等待進(jìn)程池中的所有進(jìn)程執(zhí)行完畢
- p.close()是關(guān)掉進(jìn)程池子，是不再向里面添加進(jìn)程了，對(duì)Pool對(duì)象調(diào)用join()方法會(huì)等待所有子進(jìn)程執(zhí)行完畢，調(diào)用join()之前必須先調(diào)用close()，調(diào)用close()之后就不能繼續(xù)添加新的Process了。
5、current_process()的使用
示例代碼：
from multiprocessing import Process, current_process
import time
import random
lst = []
def task(i):
print(current_process().name, i, 'start...') # current_process().name輸出進(jìn)程的名字
time.sleep(random.randint(1, 4))
lst.append(i)
print(lst)
print(current_process().name, i, 'end.....')
if __name__ == "__main__":
p_lst = []
for i in range(4):
p = Process(target=task, args=(i, ))
p_lst.append(p)
p.start()
for p in p_lst:
p.join() # 阻塞當(dāng)前進(jìn)程，直到子進(jìn)程全部退出
print("main end.......")運(yùn)行結(jié)果：
6、進(jìn)程池
進(jìn)程池里有固定數(shù)量的進(jìn)程，每次執(zhí)行任務(wù)時(shí)都從進(jìn)程池中取出一個(gè)空閑進(jìn)程來(lái)執(zhí)行，如果任務(wù)數(shù)量超過(guò)進(jìn)程池中進(jìn)程數(shù)量，那么就等待已經(jīng)在執(zhí)行的任務(wù)結(jié)束之后，有進(jìn)程空閑之后再執(zhí)行，也就是說(shuō)，同一時(shí)間，只有固定數(shù)量的進(jìn)程在執(zhí)行，這樣對(duì)操作系統(tǒng)得壓力也不會(huì)太大，效率也得到保證。
示例代碼：
from multiprocessing import Pool, current_process
import time
import random
lst = []
def task(i):
print(current_process().name, i, 'start...')
time.sleep(random.randint(1, 5))
lst.append(i)
print(lst)
print(current_process().name, i, 'end.....')
if __name__ == "__main__":
p = Pool(processes=3, maxtasksperchild=3)
for i in range(10):
p.apply_async(func=task, args=(i,)) # 進(jìn)程池接收任務(wù)
p.close() # 關(guān)閉進(jìn)程池 ==》 不接受任務(wù)
p.join() # 等待子進(jìn)程執(zhí)行完畢，父進(jìn)程再執(zhí)行
print("end.............")運(yùn)行結(jié)果：
示例代碼： 【同步執(zhí)行】
import os
import time
import random
from multiprocessing import Pool
def func1(n):
print('任務(wù){(diào)}開(kāi)始執(zhí)行，進(jìn)程為：{}'.format(n, os.getpid()))
time.sleep(random.randint(1, 4))
print('任務(wù){(diào)}結(jié)束執(zhí)行，進(jìn)程為：{}'.format(n, os.getpid()))
if __name__ == '__main__':
# c創(chuàng)建一個(gè)進(jìn)程池，里面有三個(gè)進(jìn)程
p = Pool(3)
for i in range(10):
res = p.apply(func1, args=(i,))運(yùn)行結(jié)果：
示例代碼： 【異步執(zhí)行】
import os
import time
import random
from multiprocessing import Pool
def func1(n):
print('任務(wù){(diào)}開(kāi)始執(zhí)行，進(jìn)程為：{}'.format(n, os.getpid()))
time.sleep(random.randint(1, 4))
print('任務(wù){(diào)}結(jié)束執(zhí)行，進(jìn)程為：{}'.format(n, os.getpid()))
if __name__ == '__main__':
# c創(chuàng)建一個(gè)進(jìn)程池，里面有三個(gè)進(jìn)程
p = Pool(3)
for i in range(5):
res = p.apply_async(func1, args=(i,))
p.close() # 一定要關(guān)閉
p.join() # 一定要使用join，不然進(jìn)程池里的進(jìn)程沒(méi)來(lái)得及執(zhí)行，主進(jìn)程結(jié)束了，子進(jìn)程也都跟著結(jié)束。運(yùn)行結(jié)果：
7、進(jìn)程共享變量
共享變量不適用于多進(jìn)程，進(jìn)程間的變量是互相隔離的，子進(jìn)程的全局變量是完全復(fù)制一份父進(jìn)程的數(shù)據(jù)，對(duì)子進(jìn)程的全局變量修改完全影響不到其他進(jìn)程的全局變量。
示例代碼：
import time
from multiprocessing import Process
def producer(a):
a += 1
time.sleep(2)
def consumer(a):
time.sleep(3)
data = a
print(data)
if __name__ == "__main__":
a = 1
my_producer = Process(target=producer, args=(a, ))
my_consumer = Process(target=consumer, args=(a, ))
my_producer.start()
my_consumer.start()
my_producer.join()
my_consumer.join()
# 輸出結(jié)果為1運(yùn)行結(jié)果：
示例代碼： 【進(jìn)程之間的變量是無(wú)法共享的，即使是全局變量也是不能共享的】
from multiprocessing import Process
import os
def func():
global n
n = 10
print('子進(jìn)程pid:{},n:{}'.format(os.getppid(), n))
if __name__ == '__main__':
n = 100
print('主進(jìn)程pid:{},n:{}'.format(os.getppid(), n))
p = Process(target=func)
p.start()
p.join()
print('主進(jìn)程中輸出n：{}'.format(n))運(yùn)行結(jié)果：
8、管道Pipe(兩進(jìn)程間的通信優(yōu)先考慮)
Pipe([duplex]):在進(jìn)程之間創(chuàng)建一條管道，并返回元組（conn1,conn2）,其中conn1，conn2表示管道兩端的連接對(duì)象，強(qiáng)調(diào)一點(diǎn)：必須在產(chǎn)生Process對(duì)象之前產(chǎn)生管道。dumplex:默認(rèn)管道是全雙工的，如果將duplex射成False，conn1只能用于接收，conn2只能用于發(fā)送。
- conn1.recv():接收conn2.send(obj)發(fā)送的對(duì)象。如果沒(méi)有消息可接收，recv方法會(huì)一直阻塞。如果連接的另外一端已經(jīng)關(guān)閉，那么recv方法會(huì)拋出EOFError。
- conn1.send(obj):通過(guò)連接發(fā)送對(duì)象。obj是與序列化兼容的任意對(duì)象
- conn1.close():關(guān)閉連接。如果conn1被垃圾回收，將自動(dòng)調(diào)用此方法
- conn1.fileno():返回連接使用的整數(shù)文件描述符
- conn1.poll([timeout]):如果連接上的數(shù)據(jù)可用，返回True。timeout指定等待的最長(zhǎng)時(shí)限。如果省略此參數(shù)，方法將立即返回結(jié)果。如果將timeout射成None，操作將無(wú)限期地等待數(shù)據(jù)到達(dá)。
- conn1.recv_bytes([maxlength]):接收c.send_bytes()方法發(fā)送的一條完整的字節(jié)消息。maxlength指定要接收的最大字節(jié)數(shù)。如果進(jìn)入的消息，超過(guò)了這個(gè)最大值，將引發(fā)IOError異常，并且在連接上無(wú)法進(jìn)行進(jìn)一步讀取。如果連接的另外一端已經(jīng)關(guān)閉，再也不存在任何數(shù)據(jù)，將引發(fā)EOFError異常。
- conn.send_bytes(buffer [, offset [, size]])：通過(guò)連接發(fā)送字節(jié)數(shù)據(jù)緩沖區(qū)，buffer是支持緩沖區(qū)接口的任意對(duì)象，offset是緩沖區(qū)中的字節(jié)偏移量，而size是要發(fā)送字節(jié)數(shù)。結(jié)果數(shù)據(jù)以單條消息的形式發(fā)出，然后調(diào)用c.recv_bytes()函數(shù)進(jìn)行接收
- conn1.recv_bytes_into(buffer [, offset]):接收一條完整的字節(jié)消息，并把它保存在buffer對(duì)象中，該對(duì)象支持可寫(xiě)入的緩沖區(qū)接口（即bytearray對(duì)象或類似的對(duì)象）。offset指定緩沖區(qū)中放置消息處的字節(jié)位移。返回值是收到的字節(jié)數(shù)。如果消息長(zhǎng)度大于可用的緩沖區(qū)空間，將引發(fā)BufferTooShort異常。
示例代碼：
import time
from multiprocessing import Process, Queue, Pool, Manager, Pipe
def producer(pipe):
pipe.send("a")
time.sleep(3)
print(pipe.recv())
def consumer(pipe):
time.sleep(2)
data = pipe.recv()
pipe.send("b")
print(data)
if __name__ == "__main__":
# Pipe實(shí)現(xiàn)兩進(jìn)程間通信
s_pipe, r_pipe = Pipe()
pool = Pool()
pool.apply_async(producer, args=(s_pipe, ))
pool.apply_async(consumer, args=(r_pipe, ))
pool.close()
pool.join()運(yùn)行結(jié)果：
示例代碼：
from multiprocessing import Process, Pipe
def f(conn):
conn.send('主進(jìn)程，你好呀！') # 發(fā)送數(shù)據(jù)給主進(jìn)程
print('子進(jìn)程收到主進(jìn)程發(fā)來(lái)的數(shù)據(jù)：{}'.format(conn.recv()))
conn.close() # 關(guān)閉
if __name__ == '__main__':
# Pipe是一個(gè)函數(shù)，返回的是一個(gè)元組
parent_conn, child_conn = Pipe()
# 創(chuàng)建一個(gè)子進(jìn)程
p = Process(target=f, args=(child_conn,))
p.start()
print("主進(jìn)程收到子進(jìn)程發(fā)來(lái)的數(shù)據(jù)：{}".format(parent_conn.recv()))
parent_conn.send('子進(jìn)程，你好?。?) # 發(fā)送數(shù)據(jù)給子進(jìn)程
p.join()運(yùn)行結(jié)果：
9、進(jìn)程之間的同步控制
9.1 進(jìn)程鎖：Lock()
當(dāng)多個(gè)進(jìn)程對(duì)同一資源進(jìn)行IO操作時(shí)，需要對(duì)資源“上鎖”，否則會(huì)出現(xiàn)意外結(jié)果。上鎖之后，同一件就只能有一個(gè)進(jìn)程運(yùn)行上鎖的代碼塊。例如有一個(gè)txt文件，里面內(nèi)容是一個(gè)數(shù)字10，我們要用多進(jìn)程去讀取這個(gè)文件的值，然后每讀一次，讓txt中的這個(gè)數(shù)字減1，不加鎖時(shí)代碼如下：
import time
import os
from multiprocessing import Process
from multiprocessing import Lock
def func():
if os.path.exists('num.txt'):
with open('num.txt', 'r') as rf:
num = int(rf.read())
num -= 1
time.sleep(1)
with open('num.txt', 'w') as wf:
wf.write(str(num))
else:
with open('num.txt', 'w') as wf:
wf.write('10')
if __name__ == '__main__':
print("主進(jìn)程開(kāi)始運(yùn)行……")
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('最后結(jié)果為：{}'.format(num))
print("主進(jìn)程結(jié)束運(yùn)行……" )運(yùn)行結(jié)果：
雖然用了10個(gè)進(jìn)程讀取并修改txt文件，但最后的值卻不是1。這正是多進(jìn)程共同訪問(wèn)資源造成混亂造成的。要達(dá)到預(yù)期結(jié)果，就要給資源上鎖：
import time
import os
from multiprocessing import Process
from multiprocessing import Lock
def 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('10')
if __name__ == '__main__':
print("主進(jìn)程開(kāi)始運(yùn)行……")
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('最后結(jié)果為：{}'.format(num))
print("主進(jìn)程結(jié)束運(yùn)行……")運(yùn)行結(jié)果：
果然，用了進(jìn)程鎖之后獲得了預(yù)料中的結(jié)果。但是，如果你運(yùn)行了上面兩塊代碼你就會(huì)發(fā)現(xiàn)，加了鎖之后，程序明顯變慢了很多，因?yàn)槌绦虺闪舜械牧耍?dāng)然好處是數(shù)據(jù)安全有保證。
9.2 信號(hào)量：Semaphore
鎖同時(shí)只允許一個(gè)線程更改數(shù)據(jù)，而信號(hào)量是同時(shí)允許一定數(shù)量的進(jìn)程更改數(shù)據(jù) 。假如有一下應(yīng)用場(chǎng)景：有10個(gè)人吃飯，但只有一張餐桌，只允許做3個(gè)人，沒(méi)上桌的人不允許吃飯，已上桌吃完飯離座之后，下面的人才能搶占桌子繼續(xù)吃飯，如果不用信號(hào)量，肯定是10人一窩蜂一起吃飯：
from multiprocessing import Process
import time
import random
def fun(i):
print('{}號(hào)顧客上座，開(kāi)始吃飯'.format(i))
time.sleep(random.randint(3, 8))
print('{}號(hào)顧客吃完飯了，離座'.format(i))
if __name__ == '__main__':
for i in range(10):
p = Process(target=fun, args=(i,))
p.start()運(yùn)行結(jié)果：
用了信號(hào)量，實(shí)現(xiàn)了輪流吃飯，每次只有3個(gè)人吃飯：
示例代碼：
from multiprocessing import Process
import time
import random
from multiprocessing import Semaphore
def fun(i , sem):
sem.acquire()
print('{}號(hào)顧客上座，開(kāi)始吃飯'.format(i))
time.sleep(random.randint(3, 8))
print('{}號(hào)顧客吃完飯了，離座'.format(i))
sem.release()
if __name__ == '__main__':
sem = Semaphore(3)
for i in range(10):
p = Process(target=fun, args=(i,sem))
p.start()運(yùn)行結(jié)果：
事實(shí)上，Semaphore的作用也類似于鎖，只不過(guò)在鎖機(jī)制上添加了一個(gè)計(jì)數(shù)器，允許多個(gè)人擁有“鑰匙”。
9.3 事件：Event
python進(jìn)程的事件用于主進(jìn)程控制其他子進(jìn)程的執(zhí)行，Event類有如下幾個(gè)主要方法：
1）wait() 插入在進(jìn)程中插入一個(gè)標(biāo)記（flag）默認(rèn)為 False，當(dāng) flag為False時(shí)，程序會(huì)停止運(yùn)行進(jìn)入阻塞狀態(tài)；
2）set() 使flag為T(mén)rue，程序會(huì)進(jìn)入非阻塞狀態(tài)
3）clear() 使flag為False，程序會(huì)停止運(yùn)行，進(jìn)入阻塞狀態(tài)
4）is_set() 判斷flag 是否為T(mén)rue，是的話返回True，不是則返回False
有如下需求：獲取當(dāng)前時(shí)間的秒數(shù)的個(gè)位數(shù)，如果小于5，則設(shè)置子進(jìn)程阻塞，如果大于5則設(shè)置子進(jìn)程非阻塞。代碼如下：
from multiprocessing import Event, Processimport timefrom datetime import datetimedef func(e): print('子進(jìn)程：開(kāi)始運(yùn)行……') while True: print('子進(jìn)程：現(xiàn)在事件秒數(shù)是{}'.format(datetime.now().second)) e.wait() # 阻塞等待信號(hào) 這里插入了一個(gè)flag 默認(rèn)為 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]) # 獲取當(dāng)前秒數(shù)的個(gè)位數(shù) if s < 5: print('子進(jìn)程進(jìn)入阻塞狀態(tài)') e.clear() # 使插入的flag為False 進(jìn)程進(jìn)入阻塞狀態(tài) else: print('子進(jìn)程取消阻塞狀態(tài)') e.set() # 使插入的flag為T(mén)rue，進(jìn)程進(jìn)入非阻塞狀態(tài) time.sleep(1) e.set() time.sleep(3) p.terminate() print("主進(jìn)程運(yùn)行結(jié)束……")運(yùn)行結(jié)果：
10、內(nèi)置線程池
示例代碼：
import time
import os
import random
from multiprocessing.pool import ThreadPool
def task():
print(f'開(kāi)始執(zhí)行任務(wù)：{os.getpid()}')
time.sleep(random.randint(0, 5))
print(f"執(zhí)行任務(wù)結(jié)束：{os.getpid()}")
if __name__ == '__main__':
pool = ThreadPool(2)
for i in range(5):
pool.apply_async(task)
pool.close()
pool.join()運(yùn)行結(jié)果：
到此這篇關(guān)于關(guān)于python多進(jìn)程中的常用方法詳解的文章就介紹到這了,更多相關(guān)python多進(jìn)程常用方法內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Python進(jìn)行MD5加密的實(shí)現(xiàn)全流程
MD5是一種廣泛使用的哈希算法,可將任意長(zhǎng)度的輸入數(shù)據(jù)轉(zhuǎn)換為固定長(zhǎng)度的128位哈希值,盡管其安全性已不足以應(yīng)對(duì)現(xiàn)代密碼學(xué)需求,但在數(shù)據(jù)完整性校驗(yàn)、文件防篡改檢測(cè)等場(chǎng)景中仍具實(shí)用價(jià)值,本文將詳細(xì)解析Python中實(shí)現(xiàn)MD5加密的全流程,需要的朋友可以參考下2025-09-09
利用Python將多張圖片合成視頻的實(shí)現(xiàn)
這篇文章主要介紹了利用Python將多張圖片合成視頻的實(shí)現(xiàn)，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2020-11-11
Python實(shí)現(xiàn)抓取網(wǎng)頁(yè)生成Excel文件的方法示例
這篇文章主要介紹了Python實(shí)現(xiàn)抓取網(wǎng)頁(yè)生成Excel文件的方法,涉及PyQuery模塊的使用及Excel文件相關(guān)操作技巧,需要的朋友可以參考下2017-08-08
這篇文章主要為大家介紹了Python內(nèi)建類型bytes的深入理解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-05-05
列表是Python中最常用的數(shù)據(jù)結(jié)構(gòu)之一,經(jīng)常需要對(duì)列表進(jìn)行切分操作,本文主要介紹了python中列表list切分的實(shí)現(xiàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2025-03-03
這篇文章主要介紹在python的re模塊中怎樣應(yīng)用正則表達(dá)式，文中有相關(guān)的代碼示例，具有一定的參考價(jià)值,需要的朋友可以參考下2023-06-06
下面小編就為大家分享一篇python列表的增刪改查實(shí)例代碼，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2018-01-01
Python實(shí)現(xiàn)智能合并多個(gè)Excel
在日常辦公場(chǎng)景中,Excel文件合并是數(shù)據(jù)分析人員,財(cái)務(wù)人員和行政人員經(jīng)常面臨的高頻痛點(diǎn),下面我們來(lái)看看如何使用Python實(shí)現(xiàn)智能合并多個(gè)Excel吧2025-04-04![Python筆記之a(chǎn) = [0]*x格式的含義及說(shuō)明](http://img.jbzj.com/images/xgimg/bcimg8.png)
Python筆記之a(chǎn) = [0]*x格式的含義及說(shuō)明
這篇文章主要介紹了Python筆記之a(chǎn) = [0]*x格式的含義及說(shuō)明，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-05-05
Numpy庫(kù)是Python中的一個(gè)科學(xué)計(jì)算庫(kù),本文主要介紹了ndarray的基本操作、?ndarray運(yùn)算等各種Numpy庫(kù)的超詳細(xì)教程,需要的朋友可以參考下2022-04-04