我是python的新手,我具有為數(shù)據(jù)計(jì)算特征然后返回要處理并寫入文件的列表的功能…我正在使用Pool進(jìn)行計(jì)算,然后使用回調(diào)函數(shù)來寫入文件,但是沒有調(diào)用回調(diào)函數(shù),我已經(jīng)在其中添加了一些print語句,但是絕對沒有調(diào)用它.

我的代碼如下所示：

def write_arrow_format(results):

print("writer called")

results[1].to_csv("../data/model_data/feature-"+results[2],sep='',encoding='utf-8')

with open('../data/model_data/arow-'+results[2],'w') as f:

for dic in results[0]:

feature_list=[]

print(dic)

beginLine=True

for key,value in dic.items():

if(beginLine):

feature_list.append(str(value))

beginLine=False

else:

feature_list.append(str(key)+":"+str(value))

feature_line=" ".join(feature_list)

f.write(feature_line+"

")

def generate_features(users,impressions,interactions,items,filename):

#some processing

return [result1,result2,filename]

if __name__=="__main__":

pool=mp.Pool(mp.cpu_count()-1)

for i in range(interval):

if i==interval:

pool.apply_async(generate_features,(users[begin:],impressions,interactions,items,str(i)),callback=write_arrow_format)

else:

pool.apply_async(generate_features,(users[begin:begin+interval],impressions,interactions,items,str(i)),callback=write_arrow_format)

begin=begin+interval

pool.close()

pool.join()

最佳答案

從您的帖子中還不清楚,generate_features返回的列表中包含什么.但是,如果結(jié)果1,結(jié)果2或文件名中的任何一個(gè)不可序列化,則由于某種原因,多處理庫將不會(huì)調(diào)用回調(diào)函數(shù),并且將無法靜默地執(zhí)行此操作.我認(rèn)為這是因?yàn)槎嗵幚韼靽L試在子進(jìn)程和父進(jìn)程之間來回傳遞對象之前對其進(jìn)行酸洗.如果您返回的內(nèi)容不是“可修復(fù)的”(即不可序列化),則不會(huì)調(diào)用該回調(diào).

我自己遇到了這個(gè)錯(cuò)誤,事實(shí)證明這是一個(gè)給我麻煩的logger對象的實(shí)例.這是一些示例代碼來重現(xiàn)我的問題：

import multiprocessing as mp

import logging

def bad_test_func(ii):

print('Calling bad function with arg %i'%ii)

name = "file_%i.log"%ii

logging.basicConfig(filename=name,level=logging.DEBUG)

if ii < 4:

log = logging.getLogger()

else:

log = "Test log %i"%ii

return log

def good_test_func(ii):

print('Calling good function with arg %i'%ii)

instance = ('hello', 'world', ii)

return instance

def pool_test(func):

def callback(item):

print('This is the callback')

print('I have been given the following item: ')

print(item)

num_processes = 3

pool = mp.Pool(processes = num_processes)

results = []

for i in range(5):

res = pool.apply_async(func, (i,), callback=callback)

results.append(res)

pool.close()

pool.join()

def main():

print('#'*30)

print('Calling pool test with bad function')

print('#'*30)

pool_test(bad_test_func)

print('#'*30)

print('Calling pool test with good function')

print('#'*30)

pool_test(good_test_func)

if __name__ == '__main__':

main()

希望這會(huì)有所幫助,并為您指明正確的方向.

總結(jié)

以上是生活随笔為你收集整理的python apply_async函数_Python-未调用apply_async回调函数的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。