關(guān)于python多進(jìn)程中的常用方法詳解

更新時間：2023年07月26日 09:03:45 作者：IT之一小佬

這篇文章主要介紹了關(guān)于python多進(jìn)程中的常用方法詳解,python中的多線程其實并不是真正的多線程，如果想要充分地使用多核CPU資源，在python中大部分情況需要使用多進(jìn)程,需要的朋友可以參考下

1、進(jìn)程間的通信
2、用 Queue()實現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞
3、Queue 隊列實現(xiàn)進(jìn)程間通信
4、Manage()的使用
5、current_process()的使用
6、進(jìn)程池
7、進(jìn)程共享變量
8、管道Pipe(兩進(jìn)程間的通信優(yōu)先考慮)
9、進(jìn)程之間的同步控制

9.1 進(jìn)程鎖：Lock()
9.2 信號量：Semaphore
9.3 事件：Event

10、內(nèi)置線程池

1、進(jìn)程間的通信

全局變量在多個進(jìn)程中不共享資源，進(jìn)程之間的數(shù)據(jù)是獨立的，默認(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)程開始運(yùn)行...")
    p1 = Process(target=task1)
    p2 = Process(target=task2)
    p1.start()
    p2.start()
    p1.join()
    p2.join()

運(yùn)行結(jié)果：

2、用 Queue()實現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞

Queue 是多進(jìn)程安全的隊列，可以使用 Queue 實現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞。

put 方法用以插入數(shù)據(jù)到隊列中， put 方法還有兩個可選參數(shù)： blocked 和 timeout。如果 blocked 為 True（默認(rèn)值），并且 timeout 為正值，該方法會阻塞 timeout 指定的時間，直到該隊列有剩余的空間。如果超時，會拋出 Queue.full 異常。如果 blocked 為 False，但該 Queue 已滿，會立即拋出 Queue.full 異常。

get 方法可以從隊列讀取并且刪除一個元素。同樣， get 方法有兩個可選參數(shù)： blocked和 timeout。如果 blocked 為 True（默認(rèn)值），并且 timeout 為正值，那么在等待時間內(nèi)沒有取到任何元素，會拋出 Queue.Empty 異常。如果 blocked 為 False，有兩種情況存在，如果Queue 有一個值可用，則立即返回該值，否則，如果隊列為空，則立即拋出Queue.Empty 異常

示例代碼：

from multiprocessing import Queue
q = Queue(3)
q.put('msg1')
q.put('msg2')
print('消息隊列是否已滿：', q.full())
q.put('msg3')
print('消息隊列是否已滿：', q.full())
# q.put('msg4')  # 以為消息隊列已經(jīng)滿了，需要直接寫入需要等待，如果超時會拋出異常
# 寫入數(shù)據(jù)時先判斷，判斷隊列是否已滿
if not q.full():
    q.put('msg4')
# 同理，取消息時可以先判斷隊列是否有數(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)程{}往隊列中存了一個數(shù)據(jù)：{}'.format(os.getpid(), info))
# 向queue中輸出數(shù)據(jù)的函數(shù)
def outputQ(queue):
    info = queue.get()
    print('進(jìn)程{}從隊列中取出一個數(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 隊列實現(xiàn)進(jìn)程間通信

示例代碼：

import time
from multiprocessing import Queue, Process
def write_data(q):
    # 將列表元素寫入到隊列中
    for i in ['aa', 'bb', 'cc', 'dd']:
        print('開始寫入值%s' %i)
        q.put(i)
        time.sleep(1)
def read_data(q):
    print("開始讀取數(shù)據(jù)...")
    while True:
        if not q.empty():
            print("讀取到數(shù)據(jù)：", q.get())
            time.sleep(1)
        else:
            break
if __name__ == '__main__':
    # 創(chuàng)建隊列
    q = Queue()
    # 創(chuàng)建進(jìn)程
    qw = Process(target=write_data, args=(q, ))
    qr = Process(target=read_data, args=(q, ))
    # 啟動進(jìn)程
    qw.start()
    qr.start()
    qw.join()
    qr.join()

運(yùn)行結(jié)果：

4、Manage()的使用

如果使用 Pool 創(chuàng)建進(jìn)程，就需要使用 multiprocessing.Manager()中的 Queue()來完成進(jìn)程間的通信，而不是 multiprocessing.Queue()，否則會拋出異常。

示例代碼：

import time
from multiprocessing import Manager, Pool, Queue
def write_data(q):
    # 將列表元素寫入到隊列中
    for i in ['aa', 'bb', 'cc', 'dd']:
        print('開始寫入值%s' %i)
        q.put(i)
        time.sleep(1)
def read_data(q):
    print("開始讀取數(shù)據(jù)...")
    while True:
        if not q.empty():
            print("讀取到數(shù)據(jù)：", q.get())
            time.sleep(1)
        else:
            break
if __name__ == '__main__':
    # 創(chuàng)建隊列
    q = Manager().Queue()
    # q = Queue()  # 直接這樣使用Queue()會報錯
    # 創(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
# 定義了一個foo函數(shù),接收一個字典和一個列表
def foo(dic, lst):
    # 字典和列表都放進(jìn)程ID
    dic[os.getpid()] = os.getpid()
    lst.append(os.getpid())
if __name__ == '__main__':
    # 生成Manager對象
    manager = Manager()
    dic = manager.dict()
    print(dic)
    lst = manager.list(range(3))
    print(lst)
    # 10個進(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ù)時，會一直處于等待狀態(tài)
        print(f"當(dāng)前進(jìn)程為：{current_process().name}, 隊列獲取數(shù)據(jù)為：{data},隊列剩余數(shù)據(jù)為：{queue.qsize()}個！")
        time.sleep(0.01)
if __name__ == '__main__':
    print(f"主進(jìn)程{current_process().name}開始執(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)用，否則會報ValueError: Pool is still running錯誤
    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)用，否則會報ValueError: Pool is still running錯誤
p.join() # 等待進(jìn)程池中的所有進(jìn)程執(zhí)行完畢
p.close()是關(guān)掉進(jìn)程池子，是不再向里面添加進(jìn)程了，對Pool對象調(diào)用join()方法會等待所有子進(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ù)時都從進(jìn)程池中取出一個空閑進(jìn)程來執(zhí)行，如果任務(wù)數(shù)量超過進(jìn)程池中進(jìn)程數(shù)量，那么就等待已經(jīng)在執(zhí)行的任務(wù)結(jié)束之后，有進(jìn)程空閑之后再執(zhí)行，也就是說，同一時間，只有固定數(shù)量的進(jìn)程在執(zhí)行，這樣對操作系統(tǒng)得壓力也不會太大，效率也得到保證。

示例代碼：

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)}開始執(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)建一個進(jìn)程池，里面有三個進(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)}開始執(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)建一個進(jìn)程池，里面有三個進(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)程沒來得及執(zhí)行，主進(jìn)程結(jié)束了，子進(jìn)程也都跟著結(jié)束。

運(yùn)行結(jié)果：

7、進(jìn)程共享變量

共享變量不適用于多進(jìn)程，進(jìn)程間的變量是互相隔離的，子進(jìn)程的全局變量是完全復(fù)制一份父進(jìn)程的數(shù)據(jù)，對子進(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)程之間的變量是無法共享的，即使是全局變量也是不能共享的】

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表示管道兩端的連接對象，強(qiáng)調(diào)一點：必須在產(chǎn)生Process對象之前產(chǎn)生管道。dumplex:默認(rèn)管道是全雙工的，如果將duplex射成False，conn1只能用于接收，conn2只能用于發(fā)送。

conn1.recv():接收conn2.send(obj)發(fā)送的對象。如果沒有消息可接收，recv方法會一直阻塞。如果連接的另外一端已經(jīng)關(guān)閉，那么recv方法會拋出EOFError。
conn1.send(obj):通過連接發(fā)送對象。obj是與序列化兼容的任意對象
conn1.close():關(guān)閉連接。如果conn1被垃圾回收，將自動調(diào)用此方法
conn1.fileno():返回連接使用的整數(shù)文件描述符
conn1.poll([timeout]):如果連接上的數(shù)據(jù)可用，返回True。timeout指定等待的最長時限。如果省略此參數(shù)，方法將立即返回結(jié)果。如果將timeout射成None，操作將無限期地等待數(shù)據(jù)到達(dá)。
conn1.recv_bytes([maxlength]):接收c.send_bytes()方法發(fā)送的一條完整的字節(jié)消息。maxlength指定要接收的最大字節(jié)數(shù)。如果進(jìn)入的消息，超過了這個最大值，將引發(fā)IOError異常，并且在連接上無法進(jìn)行進(jìn)一步讀取。如果連接的另外一端已經(jīng)關(guān)閉，再也不存在任何數(shù)據(jù)，將引發(fā)EOFError異常。
conn.send_bytes(buffer [, offset [, size]])：通過連接發(fā)送字節(jié)數(shù)據(jù)緩沖區(qū)，buffer是支持緩沖區(qū)接口的任意對象，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對象中，該對象支持可寫入的緩沖區(qū)接口（即bytearray對象或類似的對象）。offset指定緩沖區(qū)中放置消息處的字節(jié)位移。返回值是收到的字節(jié)數(shù)。如果消息長度大于可用的緩沖區(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實現(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ā)來的數(shù)據(jù)：{}'.format(conn.recv()))
    conn.close()  # 關(guān)閉
if __name__ == '__main__':
    # Pipe是一個函數(shù)，返回的是一個元組
    parent_conn, child_conn = Pipe()
    # 創(chuàng)建一個子進(jìn)程
    p = Process(target=f, args=(child_conn,))
    p.start()
    print("主進(jìn)程收到子進(jìn)程發(fā)來的數(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)多個進(jìn)程對同一資源進(jìn)行IO操作時，需要對資源“上鎖”，否則會出現(xiàn)意外結(jié)果。上鎖之后，同一件就只能有一個進(jìn)程運(yùn)行上鎖的代碼塊。例如有一個txt文件，里面內(nèi)容是一個數(shù)字10，我們要用多進(jìn)程去讀取這個文件的值，然后每讀一次，讓txt中的這個數(shù)字減1，不加鎖時代碼如下：

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)程開始運(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個進(jìn)程讀取并修改txt文件，但最后的值卻不是1。這正是多進(jì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)程開始運(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)行了上面兩塊代碼你就會發(fā)現(xiàn)，加了鎖之后，程序明顯變慢了很多，因為程序成了串行的了，當(dāng)然好處是數(shù)據(jù)安全有保證。

9.2 信號量：Semaphore

鎖同時只允許一個線程更改數(shù)據(jù)，而信號量是同時允許一定數(shù)量的進(jìn)程更改數(shù)據(jù) 。假如有一下應(yīng)用場景：有10個人吃飯，但只有一張餐桌，只允許做3個人，沒上桌的人不允許吃飯，已上桌吃完飯離座之后，下面的人才能搶占桌子繼續(xù)吃飯，如果不用信號量，肯定是10人一窩蜂一起吃飯：

from multiprocessing import Process
import time
import random
def fun(i):
    print('{}號顧客上座，開始吃飯'.format(i))
    time.sleep(random.randint(3, 8))
    print('{}號顧客吃完飯了，離座'.format(i))
if __name__ == '__main__':
    for i in range(10):
        p = Process(target=fun, args=(i,))
        p.start()

運(yùn)行結(jié)果：

用了信號量，實現(xiàn)了輪流吃飯，每次只有3個人吃飯：

示例代碼：

from multiprocessing import Process
import time
import random
from multiprocessing import Semaphore
def 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()

運(yùn)行結(jié)果：

事實上，Semaphore的作用也類似于鎖，只不過在鎖機(jī)制上添加了一個計數(shù)器，允許多個人擁有“鑰匙”。

9.3 事件：Event

python進(jìn)程的事件用于主進(jìn)程控制其他子進(jìn)程的執(zhí)行，Event類有如下幾個主要方法：

1）wait() 插入在進(jìn)程中插入一個標(biāo)記（flag）默認(rèn)為 False，當(dāng) flag為False時，程序會停止運(yùn)行進(jìn)入阻塞狀態(tài)；

2）set() 使flag為True，程序會進(jìn)入非阻塞狀態(tài)

3）clear() 使flag為False，程序會停止運(yùn)行，進(jìn)入阻塞狀態(tài)

4）is_set() 判斷flag 是否為True，是的話返回True，不是則返回False

有如下需求：獲取當(dāng)前時間的秒數(shù)的個位數(shù)，如果小于5，則設(shè)置子進(jìn)程阻塞，如果大于5則設(shè)置子進(jìn)程非阻塞。代碼如下：

from multiprocessing import Event, Processimport timefrom datetime import datetimedef func(e):    print(&#39;子進(jìn)程&#xff1a;開始運(yùn)行……&#39;)    while True:        print(&#39;子進(jìn)程&#xff1a;現(xiàn)在事件秒數(shù)是{}&#39;.format(datetime.now().second))        e.wait()  # 阻塞等待信號  這里插入了一個flag  默認(rèn)為 False        time.sleep(1)if __name__ &#61;&#61; &#39;__main__&#39;:    e &#61; Event()    p &#61; Process(target&#61;func, args&#61;(e,))    p.start()    for i in range(10):        s &#61; int(str(datetime.now().second)[-1])  # 獲取當(dāng)前秒數(shù)的個位數(shù)        if s &lt; 5:            print(&#39;子進(jìn)程進(jìn)入阻塞狀態(tài)&#39;)            e.clear()  # 使插入的flag為False 進(jìn)程進(jìn)入阻塞狀態(tài)        else:            print(&#39;子進(jìn)程取消阻塞狀態(tài)&#39;)            e.set()  # 使插入的flag為True&#xff0c;進(jìn)程進(jìn)入非阻塞狀態(tài)        time.sleep(1)    e.set()    time.sleep(3)    p.terminate()    print(&#34;主進(jìn)程運(yùn)行結(jié)束……&#34;)

運(yùn)行結(jié)果：

10、內(nèi)置線程池

示例代碼：

import time
import os
import random
from multiprocessing.pool import ThreadPool
def task():
    print(f'開始執(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)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

關(guān)于python多進(jìn)程中的常用方法詳解

目錄

1、進(jìn)程間的通信

2、用 Queue()實現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞

3、Queue 隊列實現(xiàn)進(jìn)程間通信

4、Manage()的使用

5、current_process()的使用

6、進(jìn)程池

7、進(jìn)程共享變量

8、管道Pipe(兩進(jìn)程間的通信優(yōu)先考慮)

9、進(jìn)程之間的同步控制

9.1 進(jìn)程鎖：Lock()

9.2 信號量：Semaphore

9.3 事件：Event

10、內(nèi)置線程池

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

關(guān)于python多進(jìn)程中的常用方法詳解

目錄

1、進(jìn)程間的通信

2、用 Queue()實現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞

3、Queue 隊列實現(xiàn)進(jìn)程間通信

4、Manage()的使用

5、current_process()的使用

6、進(jìn)程池

7、進(jìn)程共享變量

8、管道Pipe(兩進(jìn)程間的通信優(yōu)先考慮)

9、進(jìn)程之間的同步控制

9.1 進(jìn)程鎖：Lock()

9.2 信號量：Semaphore

9.3 事件：Event

10、內(nèi)置線程池

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

1、進(jìn)程間的通信

3、Queue 隊列實現(xiàn)進(jìn)程間通信

6、進(jìn)程池

7、進(jìn)程共享變量

8、管道Pipe(兩進(jìn)程間的通信優(yōu)先考慮)

9、進(jìn)程之間的同步控制