在收集資料（填充陣列）和決議資料（陣列操作）的行程之間共享numpy陣列，兩個行程都作為類方法

我的專案有一個帶有 Python 驅動程式的傳感器，它收集讀數并生成一個 numpy 陣列，該陣列需要完成一些統計處理并回傳一個或多個處理過的陣列。原始資料和處理后的資料都保存到檔案中。為了增加資料吞吐量，我們希望處理一組資料并讓傳感器驅動程式同時收集下一組資料。

驅動程式由兩個類組成，一個類class SensorInterface充當傳感器的主介面，另一個類DataCaptureThread的主方法運行在一個執行緒中，該執行緒不斷讀取套接字以獲取資料。在高級模塊中，對SensorInterface方法的呼叫會啟動將DataCaptureThread資料記錄到陣列中并在預設持續時間過去后將其保存到檔案中。

我的貢獻是開發資料處理器，它由一個單一的資料處理器組成，class DataHandler它應該在生成 numpy 陣列并執行其例程后接收它。陣列生成后，對驅動就不再重要了。

代碼的一個非常簡略的版本是這樣的（類實際上包含在不同的模塊中）：

class DataCaptureThread():
    def __init__(self,duration,parameters):
        self.duration = 0
        self.parameters = parameters
        self.capture_flag = False
        self.outgoing_data_open_flag = True
        self.thread = Threading.thread(target=self.read_data, args=(duration,))
        
    def read_data(self, duration):
        if self.capture_flag:
            while True:              
                #Pretend this function gets entire dataset from socket and saves file
                self.data_array = ReadSocket()
                #Wait for other process to indicate it's ready to receive array
                while not self.outgoing_data_open_flag:
                    continue
                
class SensorInterface():
    def __init__(self,parameters):        
        self.parameters = parameters
        
    def data_stream_start(self):
        #instantiate object of capture thread in this class
        self.capture_stream = DataCaptureThread()
        
    def collect_data(duration):
        
        self.capture_stream.duration = duration
        self.capture_stream.capture_flag = True
                
class DataHandler():
    def __init__(self):
        #flag to indicate that sensor process is not ready to transfer data
        self.incoming_data_ready_flag = False
        #Flag to indicate that process is currently busy with data, cannot accept more
        self.data_full = False
        self.data_array = np.array([])
        
    def run_processing(self):
        while True:
            #Wait until data is ready
            while not self.incoming_data_ready_flag:
                continue
            #set flag to indicate process is busy with data
            self.data_full = True
            #Pretend this function encapsulates all data processing and saving  
            DataProcessing(self.data_array)
            #Reset flag
            self.data_full = False
               
if __name__ == '__main__':
    import multiprocessing as mp
    #Create objects
    sensor = SensorInterface(params1)
    data_processor = DataHandler(params2)
    
    duration = 3.0   
    #Creating processes for data analysis
    #collect_proc = mp.Process(target=sensor.run_processing, args=(duration,) 
    data_proc = mp.Process(target=data_processor, args=())    
    #start and join processes
    data_proc.start()
    data_proc.join()

這幾乎是偽代碼，我可能忽略了初始化變數或犯了一些小錯誤。

outgoing_data_ready_flag標志、data_full和背后的意圖incoming_data_ready_flag是在接收器準備好之前阻止陣列從一個行程到另一個行程的傳輸。

我的問題是，如何在行程之間傳遞標志和陣列？Python 的多處理模塊對我來說是不透明的，我一直無法找到一種傳達這些資料的好方法，因為這兩個行程都是類方法，而標志/資料是類屬性。

請注意，我想避免過度更改驅動程式類的結構；它們非常復雜且耗時。我對資料處理類和主要功能有更直接的控制。

uj5u.com熱心網友回復：

分享旗幟

您可以使用multiprocessing.Value將標志存盤在共享記憶體中。這些將需要在主行程中創建并共享給存盤它們的子行程。此外，要訪問和更改它們的值，您需要使用value這些標志的引數，而不是直接將它們與不同的物件進行比較。

請記住，共享記憶體不是執行緒安全的，但在創建期間multiprocessing.Value支持關鍵字引數lock=True以啟用內部鎖的使用。代碼中的示例更改：

class DataCaptureThread():
    def __init__(self, capture_flag, outgoing_flag, duration,parameters):
        .
        .
        self.capture_flag = capture_flag
        self.outgoing_data_open_flag = outgoing_flag
        .
        
    def read_data(self, duration):
        if self.capture_flag.value:
            while True:              
                #Pretend this function gets entire dataset from socket and saves file
                self.data_array = ReadSocket()
                #Wait for other process to indicate it's ready to receive array
                while not self.outgoing_data_open_flag.value:
                    continue
                
class SensorInterface():
    .
    .
    .       
    def collect_data(duration):
        
        self.capture_stream.duration = duration
        self.capture_stream.capture_flag.value = True
                
class DataHandler():
    def __init__(self, incoming_data_ready_flag, data_full):
        #flag to indicate that sensor process is not ready to transfer data
        self.incoming_data_ready_flag = incoming_data_ready_flag
        #Flag to indicate that process is currently busy with data, cannot accept more
        self.data_full = data_full
        self.data_full.value = False
        .
        
    def run_processing(self):
        while True:
            #Wait until data is ready
            while not self.incoming_data_ready_flag.value:
                continue
            #set flag to indicate process is busy with data
            self.data_full.value = True
            #Pretend this function encapsulates all data processing and saving  
            DataProcessing(self.data_array)
            #Reset flag
            self.data_full.value = False
               
if __name__ == '__main__':
    import multiprocessing as mp
    from ctypes import c_bool
    
    # Create flags and set their initial value
    outgoing_data_ready_flag = Value(c_bool, False)
    data_full = Value(c_bool, False)
    incoming_data_ready_flag = Value(c_bool, False)
    
    #Create objects (remember to pass these flags to them as arguments!)
    .
    .
    .

共享陣列

根據傳達實際陣列的內容，很難提出最佳方法，因為您已經抽象了如何在偽代碼中獲取資料。通常，佇列是解決此問題的好方法，但如果只有一個消費者和一個生產者從一端寫入資料，您也可以使用管道。管道也不是執行緒安全的（與佇列不同），但通常比佇列快。如果您決定使用它們，它們還需要從主行程傳遞給子行程

此外，還有共享記憶體塊，您可以使用multiprocessing.shared_memory直接創建和存盤陣列。任何子行程都可以附加到此記憶體塊并訪問內容，這使其比佇列和管道快得多，但代價是它們處理起來有點復雜，我認為您正在努力避免這種情況。

編輯

關于共享陣列的方法，更多的是關于您準備在多大程度上為速度妥協的靈活性。例如，以下是您應該問自己的一些問題：

1. 一旦填充了陣列的內容，有多少不同的行程需要陣列的內容？如果它不止一個，那么使用佇列/管道會很昂貴，因為您需要為每個行程創建多個，以確保每個行程都獲得所需的資料。
2. 陣列中的內容是什么？請記住，從一個行程傳輸到另一個行程的所有資料（例如使用佇列/管道時）都需要腌制。因此，如果放入佇列中的物件相當復雜（或很大），酸洗然后解酸會增加額外的開銷。另一方面，共享記憶體通常對它可以存盤和不能存盤的內容非常嚴格（下面將詳細介紹）。
3. 我愿意花多少時間在這上面？就像我說的，佇列和管道是非常簡單的實作，你可以在半個晚上完成，然后就可以收工了。它們也非常靈活，您幾乎可以將任何可腌制的東西放在那里，如果代碼要求發生變化，您將來會更容易。另一方面，共享記憶體通常會帶來一系列令人頭疼的問題和瓶頸（執行緒安全是兩者兼而有之）。如果您不想像使用記憶體塊那樣低級，還有multiprocessing.Array的作業方式類似于mp.Value. 但是它再次限制了它可以存盤的內容（我從未使用過它，所以可能有一些我不知道的解決方法，用一粒鹽來處理）。

標籤：Python 多线程 麻木的 多处理

上一篇：多個矩陣的加權和

下一篇：Codeigniter4is_unique規則驗證在我更新記錄時拋出非唯一錯誤