我在多處理模塊上遇到了麻煩。我正在使用具有其map方法的工作人員池從大量文件中加載數據，對于每個文件，我都使用自定義函數來分析數據。每次處理文件時，我都希望更新一個計數器，以便跟蹤要處理的文件數量。這是示例代碼：def analyze_data( args ): # do something counter += 1 print counterif __name__ == '__main__': list_of_files = os.listdir(some_directory) global counter counter = 0 p = Pool() p.map(analyze_data, list_of_files)我找不到解決方案。
3 回答
慕標5832272
TA貢獻1966條經驗 獲得超4個贊
問題在于該counter變量未在您的進程之間共享：每個單獨的進程都在創建它自己的本地實例并對其進行遞增。
有關可用于在進程之間共享狀態的某些技術，請參閱文檔的本部分。在您的情況下，您可能希望Value在工作人員之間共享一個實例
這是示例的工作版本（帶有一些虛擬輸入數據）。請注意，它使用的是全局值，在實踐中我會盡量避免使用這些值：
from multiprocessing import Pool, Value
from time import sleep
counter = None
def init(args):
''' store the counter for later use '''
global counter
counter = args
def analyze_data(args):
''' increment the global counter, do something with the input '''
global counter
# += operation is not atomic, so we need to get a lock:
with counter.get_lock():
counter.value += 1
print counter.value
return args * 10
if __name__ == '__main__':
#inputs = os.listdir(some_directory)
#
# initialize a cross-process counter and the input lists
#
counter = Value('i', 0)
inputs = [1, 2, 3, 4]
#
# create the pool of workers, ensuring each one receives the counter
# as it starts.
#
p = Pool(initializer = init, initargs = (counter, ))
i = p.map_async(analyze_data, inputs, chunksize = 1)
i.wait()
print i.get()
www說
TA貢獻1775條經驗 獲得超8個贊
沒有競爭條件錯誤的計數器類：
class Counter(object):
def __init__(self):
self.val = multiprocessing.Value('i', 0)
def increment(self, n=1):
with self.val.get_lock():
self.val.value += n
@property
def value(self):
return self.val.value
添加回答
舉報
0/150
提交
取消