未调用 apply_async 回调函数

Question

我是 python 的新手，我有一个函数可以计算我的数据的特征，然后返回一个应该处理并写入文件的列表。，我正在使用 Pool 进行计算，然后使用回调函数来写入文件，但是没有调用回调函数，我已经在其中放入了一些打印语句，但它肯定没有被调用。我的代码是这样的：

def write_arrow_format(results):
print("writer called")
results[1].to_csv("../data/model_data/feature-"+results[2],sep='\t',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+"\n")


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

Answer 1

从您的帖子中看不出generate_features. 但是，如果result1, result2, 或filename中的任何一个不可序列化，那么由于某种原因，多处理库将不会调用回调函数，并且无法静默执行。我认为这是因为多处理库尝试在子进程和父进程之间来回传递对象之前对其进行腌制。如果您返回的任何内容不是“pickleable”（即不可序列化），则不会调用回调。

我自己也遇到过这个错误，结果是一个记录器对象的实例给我带来了麻烦。这是一些示例代码来重现我的问题：

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

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