快捷導(dǎo)航

python多進(jìn)程共享Array問題

更新時(shí)間：2024年09月09日 16:07:40 作者：言之兮兮

multiprocessing庫(kù)提供了Array類,允許在多個(gè)進(jìn)程間共享數(shù)組,Array在共享內(nèi)存中創(chuàng)建,各進(jìn)程可直接訪問和修改其元素,實(shí)現(xiàn)數(shù)據(jù)同步,Array支持多種數(shù)據(jù)類型,可選鎖定參數(shù)以保證數(shù)據(jù)安全

目的

可以使用multiprocessing庫(kù)中的Array來實(shí)現(xiàn)多進(jìn)程共享Array。

multiprocessing庫(kù)是一個(gè)用于實(shí)現(xiàn)多進(jìn)程編程的庫(kù)，提供了與多進(jìn)程有關(guān)的同步，通信和進(jìn)程管理等功能。

在這個(gè)庫(kù)中，有一個(gè)Array類，用于創(chuàng)建一個(gè)多進(jìn)程共享的數(shù)組。

這樣，每個(gè)進(jìn)程都可以訪問這個(gè)共享的數(shù)組，從而實(shí)現(xiàn)數(shù)據(jù)共享和同步。

基礎(chǔ)知識(shí)介紹

1.multiprocessing.Array

multiprocessing.Array是Python的多進(jìn)程庫(kù)中的一個(gè)數(shù)組對(duì)象，它用于在多個(gè)進(jìn)程之間共享數(shù)組數(shù)據(jù)。
Array類型是在共享內(nèi)存中實(shí)現(xiàn)的，每個(gè)進(jìn)程可以操作該數(shù)組的值。這是在多個(gè)進(jìn)程間同步數(shù)據(jù)的一種常用方法，因?yàn)樗鼈冊(cè)谕粌?nèi)存中運(yùn)行。
使用multiprocessing.Array，你可以創(chuàng)建一個(gè)共享的數(shù)組，并在多個(gè)進(jìn)程中讀取和修改數(shù)組的值。
通過使用該模塊，您可以避免將數(shù)組通過管道或套接字從一個(gè)進(jìn)程傳遞到另一個(gè)進(jìn)程的問題。

2.multiprocessing.Array 函數(shù)接受以下參數(shù)

typecode：字符串，表示數(shù)組元素的數(shù)據(jù)類型。可用的數(shù)據(jù)類型包括 'b'（布爾），'i'（整數(shù)），'f'（浮點(diǎn)數(shù)），'c'（單字符），'u'（Unicode字符）等。
size：整數(shù)，表示數(shù)組的大小。
lock：布爾值，指示是否應(yīng)該為數(shù)組元素創(chuàng)建鎖（鎖定）。

以上參數(shù)中，前兩個(gè)是必需的，而后一個(gè)是可選的。

python多進(jìn)程共享一維Array

舉個(gè)例子，你可以使用multiprocessing庫(kù)中的Array創(chuàng)建一個(gè)整型數(shù)組，并在兩個(gè)進(jìn)程中分別執(zhí)行讀取和寫入操作，從而實(shí)現(xiàn)多進(jìn)程共享Array。

代碼

from multiprocessing import Process, Array
import time


def print_array(arr, n):
    while True:
        for i in range(n):
            print("Process {}: {}".format(i, arr[i]))
        time.sleep(1)


test_count = 1


def set_array(arr, n):
    global test_count
    while True:
        for i in range(n):
            arr[i] = i + test_count
        time.sleep(1)
        test_count += 1
        print(test_count)


if __name__ == "__main__":
    arr = Array('i', [0, 0, 0, 0, 0, 0, 0, 0, 0])
    n = len(arr)
    process1 = Process(target=print_array, args=(arr, n))
    process2 = Process(target=set_array, args=(arr, n))

    process1.start()
    process2.start()

    process1.join()
    process2.join()

這段代碼創(chuàng)建了兩個(gè)進(jìn)程：一個(gè)打印數(shù)組的進(jìn)程（print_array）和一個(gè)設(shè)置數(shù)組的進(jìn)程（set_array）。這兩個(gè)進(jìn)程同時(shí)運(yùn)行，互不影響。

print_array 進(jìn)程通過一個(gè)循環(huán)打印數(shù)組中的每一個(gè)元素，并在每次循環(huán)之后等待 1 秒。
set_array 進(jìn)程通過一個(gè)循環(huán)設(shè)置數(shù)組中的每一個(gè)元素，并在每次循環(huán)之后等待 1 秒。它設(shè)置的數(shù)組的每一個(gè)元素的值都是經(jīng)過累加的。

通過 Array 對(duì)象，兩個(gè)進(jìn)程共享同一個(gè)數(shù)組。需要注意的是，在 Python 中多進(jìn)程之間共享數(shù)據(jù)時(shí)需要特殊處理，以保證同步與數(shù)據(jù)完整性。這里使用的 multiprocessing.Array 可以方便地共享數(shù)組。

程序的主流程是創(chuàng)建兩個(gè)進(jìn)程，啟動(dòng)它們并等待它們結(jié)束。當(dāng)兩個(gè)進(jìn)程都結(jié)束后，程序結(jié)束。

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

Process 0: 0
Process 1: 0
Process 2: 0
Process 3: 0
Process 4: 0
Process 5: 0
Process 6: 0
Process 7: 0
Process 8: 0
2
Process 0: 1
Process 1: 3
Process 2: 4
Process 3: 5
Process 4: 6
Process 5: 7
Process 6: 8
Process 7: 9
Process 8: 10
Process 0: 2
Process 1: 3
Process 2: 4
3Process 3: 5

Process 4: 6
Process 5: 7
Process 6: 8
Process 7: 9
Process 8: 10
4
Process 0: 3
Process 1: 4
Process 2: 6
Process 3: 7
Process 4: 8
Process 5: 9
Process 6: 10
Process 7: 11
Process 8: 12

python多進(jìn)程共享多維Array

在 Python 中，使用multiprocessing模塊的Array來共享多維數(shù)組在多進(jìn)程中需要用到一些特殊的技巧。

首先，你需要定義一個(gè)共享內(nèi)存，并在多個(gè)進(jìn)程中共享該內(nèi)存。

例如，如果你想要共享一個(gè)3維數(shù)組，可以定義一個(gè)二維共享內(nèi)存數(shù)組，然后再在每個(gè)進(jìn)程中訪問它，它仍然是共享的。

下面是一個(gè)示例代碼，該代碼演示了如何在兩個(gè)進(jìn)程中共享一個(gè)3維數(shù)組：

代碼

from multiprocessing import Process, Array
import time
import numpy as np


def print_array(arr, shape):
    while True:
        print("Process: \n")
        print(np.array(arr).reshape(shape))
        time.sleep(1)


def set_array(arr, shape):
    while True:
        np.random.shuffle(arr)
        time.sleep(1)
        print("Shuffled the array")


if __name__ == "__main__":
    shape = (2, 3, 4)
    arr = Array('d', np.arange(np.prod(shape)))
    process1 = Process(target=print_array, args=(arr, shape))
    process2 = Process(target=set_array, args=(arr, shape))

    process1.start()
    process2.start()

    process1.join()
    process2.join()

在上面的代碼中，我們使用了multiprocessing庫(kù)中的Array來實(shí)現(xiàn)多進(jìn)程共享一個(gè)3維數(shù)組，它們可以在兩個(gè)獨(dú)立的進(jìn)程中讀取和修改這個(gè)數(shù)組。

打印結(jié)果可能是一個(gè)不斷變化的3維數(shù)組，每隔1秒進(jìn)行重新洗牌。

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

Process:
[[[ 0. 1. 2. 3.]
[ 4. 5. 6. 7.]
[ 8. 9. 10. 11.]]

[[12. 13. 14. 15.]
[16. 17. 18. 19.]
[20. 21. 22. 23.]]]

警告信息

UserWarning: you are shuffling a 'SynchronizedArray' object which is not a subclass of 'Sequence'; `shuffle` is not guaranteed to behave correctly. E.g., non-numpy array/tensor objects with view semantics may contain duplicates after shuffling.