python apply_async函数_Python-未调用apply_async回调函数

我是python的新手,我具有为数据计算特征然后返回要处理并写入文件的列表的功能…我正在使用Pool进行计算,然后使用回调函数来写入文件,但是没有调用回调函数,我已经在其中添加了一些print语句,但是绝对没有调用它.

我的代码如下所示：

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返回的列表中包含什么.但是,如果结果1,结果2或文件名中的任何一个不可序列化,则由于某种原因,多处理库将不会调用回调函数,并且将无法静默地执行此操作.我认为这是因为多处理库尝试在子进程和父进程之间来回传递对象之前对其进行酸洗.如果您返回的内容不是“可修复的”(即不可序列化),则不会调用该回调.

我自己遇到了这个错误,事实证明这是一个给我麻烦的logger对象的实例.这是一些示例代码来重现我的问题：

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()

希望这会有所帮助,并为您指明正确的方向.

总结

以上是生活随笔为你收集整理的python apply_async函数_Python-未调用apply_async回调函数的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。