快捷導(dǎo)航

python使用multiprocessing的詳細(xì)方法

更新時(shí)間：2024年03月13日 14:25:34 作者：fakerth

multiprocessing是Python標(biāo)準(zhǔn)庫(kù)中的一個(gè)模塊,用于實(shí)現(xiàn)多進(jìn)程編程,它提供了一種簡(jiǎn)單而高效的方式來(lái)利用多核處理器的能力,通過(guò)在多個(gè)進(jìn)程中同時(shí)執(zhí)行任務(wù),加快程序的執(zhí)行速度和提高系統(tǒng)的吞吐量,這篇文章主要介紹了python使用multiprocessing,需要的朋友可以參考下

multiprocessing

multiprocessing是Python標(biāo)準(zhǔn)庫(kù)中的一個(gè)模塊，用于實(shí)現(xiàn)多進(jìn)程編程。它提供了一種簡(jiǎn)單而高效的方式來(lái)利用多核處理器的能力，通過(guò)在多個(gè)進(jìn)程中同時(shí)執(zhí)行任務(wù)，加快程序的執(zhí)行速度和提高系統(tǒng)的吞吐量。

下面是使用multiprocessing模塊的一些常見(jiàn)操作：

創(chuàng)建進(jìn)程：

使用Process類(lèi)創(chuàng)建進(jìn)程對(duì)象，指定要執(zhí)行的函數(shù)或方法。
使用Process類(lèi)的start()方法啟動(dòng)進(jìn)程。

進(jìn)程間通信：
- 使用Queue類(lèi)實(shí)現(xiàn)進(jìn)程間的隊(duì)列通信。
- 使用Pipe類(lèi)實(shí)現(xiàn)進(jìn)程間的管道通信。
- 使用共享內(nèi)存（Value和Array）實(shí)現(xiàn)進(jìn)程間的數(shù)據(jù)共享。
進(jìn)程管理：
- 使用Process類(lèi)的join()方法等待進(jìn)程結(jié)束。
- 使用Process類(lèi)的terminate()方法終止進(jìn)程。

process

?multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

參數(shù)介紹：

group默認(rèn)為None（目前未使用）
target代表調(diào)用對(duì)象，即子進(jìn)程執(zhí)行的任務(wù)
name為進(jìn)程名稱(chēng)
args調(diào)用對(duì)象的位置參數(shù)元組，args=(value1, value2, …)
kwargs調(diào)用對(duì)象的字典，kwargs={key1:value1, key2:value2, …}
daemon表示進(jìn)程是否為守護(hù)進(jìn)程，布爾值

方法介紹：

Process.start() 啟動(dòng)進(jìn)程，并調(diào)用子進(jìn)程中的run()方法
Process.run() 進(jìn)程啟動(dòng)時(shí)運(yùn)行的方法，在自定義時(shí)必須要實(shí)現(xiàn)該方法
Process.terminate() 強(qiáng)制終止進(jìn)程，不進(jìn)行清理操作，如果Process創(chuàng)建了子進(jìn)程，會(huì)導(dǎo)致該進(jìn)程變成僵尸進(jìn)程
Process.join() 阻塞進(jìn)程使主進(jìn)程等待該進(jìn)程終止
Process.kill() 與terminate()相同
Process.is_alive() 判斷進(jìn)程是否還存活，如果存活，返回True
Process.close() 關(guān)閉進(jìn)程對(duì)象，并清理資源，如果進(jìn)程仍在運(yùn)行則返回錯(cuò)誤

multiprocessing.Queue()

multiprocessing.Queue()是multiprocessing模塊中的一個(gè)類(lèi)，用于實(shí)現(xiàn)進(jìn)程間通信的隊(duì)列（Queue）。它提供了一種安全的方式，讓多個(gè)進(jìn)程之間可以共享數(shù)據(jù)。multiprocessing.Queue()類(lèi)的主要特點(diǎn)包括：

安全性：multiprocessing.Queue()是線程安全的，可以在多個(gè)進(jìn)程中同時(shí)使用，而無(wú)需擔(dān)心數(shù)據(jù)競(jìng)爭(zhēng)或不一致性問(wèn)題。
先進(jìn)先出（FIFO）：它遵循先進(jìn)先出的原則，保證了添加到隊(duì)列中的元素按照添加的順序被取出。
阻塞操作：當(dāng)隊(duì)列為空時(shí)，使用get()方法從隊(duì)列中獲取元素會(huì)阻塞進(jìn)程，直到隊(duì)列中有可用的元素。當(dāng)隊(duì)列滿(mǎn)時(shí)，使用put()方法向隊(duì)列中添加元素會(huì)阻塞進(jìn)程，直到隊(duì)列有空閑空間。

import multiprocessing
def worker(queue):
    data = queue.get()  # 從隊(duì)列中獲取數(shù)據(jù)
    # 處理數(shù)據(jù)
if __name__ == '__main__':
    queue = multiprocessing.Queue()
    process = multiprocessing.Process(target=worker, args=(queue,))
    process.start()
    queue.put(data)  # 向隊(duì)列中添加數(shù)據(jù)
    process.join()

在上面的示例中，首先創(chuàng)建了一個(gè)multiprocessing.Queue()對(duì)象，然后將該隊(duì)列對(duì)象作為參數(shù)傳遞給子進(jìn)程的worker()函數(shù)。在子進(jìn)程中，使用get()方法從隊(duì)列中獲取數(shù)據(jù)進(jìn)行處理。在主進(jìn)程中，使用put()方法向隊(duì)列中添加數(shù)據(jù)。通過(guò)使用multiprocessing.Queue()，可以讓多個(gè)進(jìn)程之間安全地傳遞數(shù)據(jù)，實(shí)現(xiàn)進(jìn)程間的通信和協(xié)作。這對(duì)于并行計(jì)算、任務(wù)分發(fā)和處理等場(chǎng)景非常有用。

拿之前的點(diǎn)點(diǎn)帶寬舉例

七個(gè)節(jié)點(diǎn)不重復(fù)取兩個(gè)，C72也就是21組，即21次循環(huán)，每次循環(huán)sleep5秒，串行就是21x5=105秒，21個(gè)線程并行5秒。

import multiprocessing
import time
import random
def get_oobw_parallel(node_names):
    results = []
    for i in range(0, len(node_names) - 1):
        for j in range(i + 1, len(node_names)):
            result = get_oobw(node_names[i], node_names[j])
            results.append(result)
    return results
def get_oobw(node_name1, node_name2):
    # 執(zhí)行 get_oobw 的邏輯
    # ...
    time.sleep(5)
    latency, bandwidth = round(random.uniform(100.0, 200.0), 4), round(random.uniform(100.0, 200.0), 4)
    result = (node_name1, node_name2, latency, bandwidth)
    return result
start_time = time.time()
node_names = ["cn1", "cn2", "cn3", "cn4", "cn5", "cn6", "cn7"]  # 填入你的節(jié)點(diǎn)名稱(chēng)列表
results = get_oobw_parallel(node_names)
for result in results:
    node_name1, node_name2, latency, bandwidth = result
    print(node_name1, node_name2, latency, bandwidth)
end_time = time.time()
# 計(jì)算執(zhí)行時(shí)間
execution_time = end_time - start_time
print("程序執(zhí)行時(shí)間：", execution_time, "秒")

import multiprocessing
import time
import random
def get_oobw_parallel(node_names):
    results = []
    processes = []
    result_queue = multiprocessing.Queue()
    for i in range(0, len(node_names) - 1):
        for j in range(i + 1, len(node_names)):
            process = multiprocessing.Process(target=get_oobw, args=(node_names[i], node_names[j], result_queue))
            process.start()
            processes.append(process)
    for process in processes:
        process.join()
    while not result_queue.empty():
        result = result_queue.get()
        results.append(result)
    return results
def get_oobw(node_name1, node_name2, result_queue):
    # 執(zhí)行 get_oobw 的邏輯
    # ...
    time.sleep(5)
    latency, bandwidth = round(random.uniform(100.0, 200.0), 4), round(random.uniform(100.0, 200.0), 4)
    result = (node_name1, node_name2, latency, bandwidth)
    result_queue.put(result)
    # return latency, bandwidth
start_time = time.time()
node_names = ["cn1", "cn2", "cn3", "cn4", "cn5", "cn6", "cn7"]  # 填入你的節(jié)點(diǎn)名稱(chēng)列表
results = get_oobw_parallel(node_names)
for result in results:
    node_name1, node_name2, latency, bandwidth = result
    print(node_name1, node_name2, latency, bandwidth)
end_time = time.time()
# 計(jì)算執(zhí)行時(shí)間
execution_time = end_time - start_time
print("程序執(zhí)行時(shí)間：", execution_time, "秒")