使用python在多处理中共享字典

r囧r小猫 2023-10-04 09:45 6阅读 0赞

在我的程序中，我需要在使用Python进行多处理的进程之间共享一个字典。我简化了代码，在此举一个例子：

import multiprocessing
    def folding (return_dict, seq):
        dis = 1
        d = 0
        ddg = 1 
        '''This is irrelevant, actually my program sends seq parameter to other extern program that returns dis, d and ddg parameters'''
        return_dict [seq] = [dis, d, ddg]
    
    seqs = ['atcgtg', 'agcgatcg', 'atcgatcgatc', atcggatcg', agctgctagct']
    manager = Manager()
    return_dict = manager.dict()
    n_cores = 3
    
    for i in range (0, len(seqs), n_cores) #n_cores is the number of cores availables in the computer, defined by the user
        subseqs = seqs[i:i + n_cores]
        processes = [Process(target=folding, args =(return_dict, seq)) for seq in subseqs]
        for p in processes:
            p.start()
        for p in processes:
            p.join()
    
    for i in retun_dict:
        print i

我希望在程序的末尾具有所有属性值的return\_dict。当我运行程序时，必须使用数千个序列的seqs列表来执行此操作，并重复很多次，有时我会得到正确的结果，但是有时（大部分时间）该程序永远不会结束，但是会重新调整任何内容错误，我不知道出了什么问题。此外，我认为这在时间上不是很有效，我想知道是否还有其他方法可以最有效，更快地执行此操作。谢谢大家！

## 最佳方法： ##

通过修复一些小的语法错误，您的代码似乎可以正常工作。

但是，我将使用[多处理池][Link 1]而不是您的自定义解决方案来一次始终运行`n_cores`进程。您的方法存在的问题是，在开始下一批之前，所有过程都需要完成。根据所需时间的可变程度`folding`，您可能会遇到瓶颈。在最坏的情况下，这意味着与单核处理相比根本无法提高速度。

而且，您的程序在Windows下会遇到严重的问题。您需要确保可以安全地导入主模块，而无需重新运行多处理代码。也就是说，你需要保护你的主要*切入点，*通过`if __name__ == '__main___'`它你可以在其他Python脚本已经看到了。这将确保仅在脚本作为解释器的*主*文件启动时才执行受保护的代码，即仅执行一次，而不是由您产生的每个新子进程启动。

这是我使用池对代码进行的稍微改动的版本：

import multiprocessing as mp
    
    
    def folding(return_dict, seq):
        dis = 1
        d = 0
        ddg = 1
        '''This is irrelevant, actually my program sends seq parameter to other extern program that returns dis, d and ddg parameters'''
        return_dict[seq] = [dis, d, ddg]
    
    
    def main():
        seqs = ['atcgtg', 'agcgatcg', 'atcgatcgatc', 'atcggatcg', 'agctgctagct']
        manager = mp.Manager()
        return_dict = manager.dict()
        n_cores = 3
    
        # created pool running maximum 3 cores
        pool = mp.Pool(n_cores)
    
        # Execute the folding task in parallel
        for seq in seqs:
            pool.apply_async(folding, args=(return_dict, seq))
    
        # Tell the pool that there are no more tasks to come and join
        pool.close()
        pool.join()
    
        # Print the results
        for i in return_dict.keys():
            print(i, return_dict[i])
    
    
    if __name__ == '__main__':
        # Protected main function
        main()

这将打印

atcgtg [1, 0, 1]
    atcgatcgatc [1, 0, 1]
    agcgatcg [1, 0, 1]
    atcggatcg [1, 0, 1]
    agctgctagct [1, 0, 1]

--------------------

### 没有共享数据的示例 ###

**编辑：**同样在您的情况下，实际上不需要共享数据结构。您可以简单地依靠[池的地图][Link 2]功能。*map*接受一个*iterable*，然后*将*其`folding`与iterable的所有元素一起调用该函数。在*map\_asnyc*上使用*map*的好处是结果按输入顺序排列。但是您需要等到所有结果都收集完毕后才能处理它们。

这是使用*map*的示例。请注意，您的函数`folding`现在返回结果，而不是将其放入共享字典中：

import multiprocessing as mp
    
    
    def folding(seq):
        dis = 1
        d = 0
        ddg = 1
        '''This is irrelevant, actually my program sends seq parameter to other extern program that returns dis, d and ddg parameters'''
        # Return results instead of using shared data
        return [dis, d, ddg]
    
    
    def main():
        seqs = ['atcgtg', 'agcgatcg', 'atcgatcgatc', 'atcggatcg', 'agctgctagct']
        n_cores = 3
    
        pool = mp.Pool(n_cores)
    
        # Use map which blocks until all results are ready:
        res = pool.map(folding, iterable=seqs)
    
        pool.close()
        pool.join()
    
        # Print the results
        # Using list(zip(..)) to print inputs next to outputs
        print(list(zip(seqs, res)))
    
    
    if __name__ == '__main__':
        main()

这一个打印

[('atcgtg', [1, 0, 1]), ('agcgatcg', [1, 0, 1]), ('atcgatcgatc', [1, 0, 1]), ('atcggatcg', [1, 0, 1]), ('agctgctagct', [1, 0, 1])]

[Link 1]: https://docs.python.org/2/library/multiprocessing.html#module-multiprocessing.pool
[Link 2]: https://docs.python.org/2/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.map