python中簡單好用的進程間數據通信模塊multiprocessing.Manager

目前開發中有遇到進程間須要共享數據的狀況. 因此研究了下multiprocessing.Manager, 主要會以dict爲例子, 說明下進程間共享(同一個父進程).

dict使用說明

import multiprocessing
# 1. 建立一個Manger對象
manager = multiprocessing.Manager()
# 2. 建立一個dict
temp_dict = manager.dict()
# 3. 建立一個測試程序
def test(idx, test_dict):
    test_dict[idx] = idx
# 4. 建立進程池進行測試
pool = multiprocessing.Pool(4)
for i in range(100):
    pool.apply_async(test, args=(i, temp_dict))
pool.close()
pool.join()
print(temp_dict)

too simple.

簡單的源碼分析

這時咱們再看一個例子

import multiprocessing
# 1. 建立一個Manger對象
manager = multiprocessing.Manager()
# 2. 建立一個dict
temp_dict = manager.dict()
temp_dict['test'] = {}
# 3. 建立一個測試程序
def test(idx, test_dict):
    test_dict['test'][idx] = idx
# 4. 建立進程池進行測試
pool = multiprocessing.Pool(4)
for i in range(100):
    pool.apply_async(test, args=(i, temp_dict))
pool.close()
pool.join()
print(temp_dict)

能夠看到輸出結果是奇怪的{'test': {}}
若是咱們簡單修改一下代碼

import multiprocessing
# 1. 建立一個Manger對象
manager = multiprocessing.Manager()
# 2. 建立一個dict
temp_dict = manager.dict()
temp_dict['test'] = {}
# 3. 建立一個測試程序
def test(idx, test_dict):
    row = test_dict['test']
    row[idx] = idx
    test_dict['test'] = row
# 4. 建立進程池進行測試
pool = multiprocessing.Pool(4)
for i in range(100):
    pool.apply_async(test, args=(i, temp_dict))
pool.close()
pool.join()
print(temp_dict)

這時輸出結果就符合預期了.

爲了瞭解這個現象背後的緣由, 我簡單去讀了一下源碼, 主要有如下幾段代碼很關鍵.

def Manager():
    '''
    Returns a manager associated with a running server process

    The managers methods such as `Lock()`, `Condition()` and `Queue()`
    can be used to create shared objects.
    '''
    from multiprocessing.managers import SyncManager
    m = SyncManager()
    m.start()
    return m
    
...
    def start(self, initializer=None, initargs=()):
        '''
        Spawn a server process for this manager object
        '''
        assert self._state.value == State.INITIAL

        if initializer is not None and not hasattr(initializer, '__call__'):
            raise TypeError('initializer must be a callable')

        # pipe over which we will retrieve address of server
        reader, writer = connection.Pipe(duplex=False)

        # spawn process which runs a server
        self._process = Process(
            target=type(self)._run_server,
            args=(self._registry, self._address, self._authkey,
                  self._serializer, writer, initializer, initargs),
            )
        ident = ':'.join(str(i) for i in self._process._identity)
        self._process.name = type(self).__name__  + '-' + ident
        self._process.start()
...

上面代碼能夠看出, 當咱們聲明瞭一個Manager對象的時候, 程序實際在其餘進程啓動了一個server服務, 這個server是阻塞的, 以此來實現進程間數據安全.
個人理解就是不一樣進程之間操做都是互斥的, 一個進程向server請求到這部分數據, 再把這部分數據修改, 返回給server, 以後server再去處理其餘進程的請求.

回到上面的奇怪現象上, 這個操做test_dict['test'][idx] = idx實際上在拉取到server上的數據後進行了修改, 但並無返回給server, 因此temp_dict的數據根本沒有變化. 在第二段正常代碼, 就至關於先向服務器請求數據, 再向服務器傳送修改後的數據. 這樣就能夠解釋這個現象了.

進程間數據安全

這個時候若是出現一種狀況, 兩個進程同時請求了一份相同的數據, 分別進行修改, 再提交到server上會怎麼樣呢? 那固然是數據產生異常. 基於此, 咱們須要Manager的另外一個對象, Lock(). 這個對象也不難理解, Manager自己就是一個server, dict跟lock都來自於這個server, 因此當你lock住的時候, 其餘進程是不能取到數據, 天然也不會出現上面那種異常狀況.

代碼示例:

import multiprocessing
# 1. 建立一個Manger對象
manager = multiprocessing.Manager()
# 2. 建立一個dict
temp_dict = manager.dict()
lock = manager.Lock()
temp_dict['test'] = {}
# 3. 建立一個測試程序
def test(idx, test_dict, lock):
    lock.acquire()
    row = test_dict['test']
    row[idx] = idx
    test_dict['test'] = row
    lock.release()
# 4. 建立進程池進行測試
pool = multiprocessing.Pool(4)
for i in range(100):
    pool.apply_async(test, args=(i, temp_dict, lock))
pool.close()
pool.join()
print(temp_dict)

切忌不要進程裏本身新建lock對象, 要使用統一的lock對象.

終わり。