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

jeyhan 發布于2019-07-31 10:12 / 2886人閱讀

摘要：目前開發中有遇到進程間需要共享數據的情況所以研究了下主要會以為例子說明下進程間共享同一個父進程使用說明創建一個對象創建一個創建一個測試程序創建進程池進行測試簡單的源碼分析這時我們再看一個例子創建一個對象創建一個創建一個測試程序創建進程池進行

目前開發中有遇到進程間需要共享數據的情況. 所以研究了下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對象.

終わり。

GPU云服務器云服務器進程間通訊 linux 進程間數據 linux進程間數據交換 linux 進程間共享數據

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/43443.html

python進階筆記【1】--- 多進程

摘要：很簡單，這個模塊實現了開辟一塊共享內存空間，就好比中的方法一樣，有興趣的同學可以去查閱。查了下資料，返回的對象控制了一個進程，可用于多進程之間的安全通信，其支持的類型有和等。有關于 multiprocessing 中共享變量的問題現在的cpu都很強大，比方我用的至強2620有24核可以同時工作，并行執行進程程序。這在計算密集型的程序是很需要的，如沙漠中的綠洲，令人重獲新生。那么，問...

Wildcard 2019-07-31 10:58 評論0 收藏0
Python進程專題7：托管對象

摘要：連接帶遠程管理器對象，該對象的地址在構造函數中支出。在當前進程中運行管理器服務器。啟動一個單的子進程，并在該子進程中啟動管理器服務器。如果無法序列號對象將引發異常。上一篇文章：Python進程專題6：共享數據與同步下一篇文章：Python進程專題8：分布集群的消息傳遞進程不支持共享對象，上面描述的創建共享值和數組，但都是指定的特殊類型，對高級的Python對象（如：字典、列表、用...

DevYK 2019-07-30 17:41 評論0 收藏0
實戰案例分享：利用Python實現多任務進程

摘要：效率高當然，對于爬蟲這種密集型任務來說，多線程和多進程影響差別并不大。對于計算密集型任務來說，的多進程相比多線程，其多核運行效率會有成倍的提升。一、進程介紹進程...

MudOnTire 2021-11-24 09:39 評論0 收藏0
python并發4：使用thread處理并發

摘要：如果某線程并未使用很多操作，它會在自己的時間片內一直占用處理器和。在中使用線程在和等大多數類系統上運行時，支持多線程編程。守護線程另一個避免使用模塊的原因是，它不支持守護線程。這一篇是Python并發的第四篇，主要介紹進程和線程的定義，Python線程和全局解釋器鎖以及Python如何使用thread模塊處理并發引言&動機考慮一下這個場景，我們有10000條數據需要處理，處理每條...

joywek 2019-07-25 11:38 評論0 收藏0
Python 的并發編程

摘要：本文最先發布在博客這篇文章將講解并發編程的基本操作。并發是指能夠多任務處理，并行則是是能夠同時多任務處理。雖然自帶了很好的類庫支持多線程進程編程，但眾所周知，因為的存在，很難做好真正的并行。本文最先發布在博客：https://blog.ihypo.net/151628... 這篇文章將講解 Python 并發編程的基本操作。并發和并行是對孿生兄弟，概念經常混淆。并發是指能夠多任務處...

happen 2019-07-30 16:29 評論0 收藏0