快捷導(dǎo)航

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

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

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

multiprocessing

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

下面是使用multiprocessing模塊的一些常見操作：

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

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

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

process

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

參數(shù)介紹：

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

方法介紹：

Process.start() 啟動進(jìn)程，并調(diào)用子進(jìn)程中的run()方法
Process.run() 進(jìn)程啟動時運(yùn)行的方法，在自定義時必須要實現(xiàn)該方法
Process.terminate() 強(qiáng)制終止進(jìn)程，不進(jìn)行清理操作，如果Process創(chuàng)建了子進(jìn)程，會導(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)程對象，并清理資源，如果進(jìn)程仍在運(yùn)行則返回錯誤

multiprocessing.Queue()

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

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

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

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

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

七個節(jié)點(diǎn)不重復(fù)取兩個，C72也就是21組，即21次循環(huán)，每次循環(huán)sleep5秒，串行就是21x5=105秒，21個線程并行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)名稱列表
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()
# 計算執(zhí)行時間
execution_time = end_time - start_time
print("程序執(zhí)行時間：", 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)名稱列表
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()
# 計算執(zhí)行時間
execution_time = end_time - start_time
print("程序執(zhí)行時間：", execution_time, "秒")