多处理池挂起且无法脱离应用程序

Question

我确信这是一个菜鸟错误，但我无法弄清楚我在多处理方面做错了什么。我有这个代码（只是坐在周围，什么都不做）

if __name__ == '__main__':
    pool = Pool(processes=4)  
    for i, x in enumerate(data): 
        pool.apply_async(new_awesome_function, (i, x))
    pool.close()
    pool.join()

数据是一个列表（[1,2,3,4,5]），我试图将列表发送到多个CPU上完成的每个项目，但是当我将工作命令包装到一个函数中并发送此代码，它不会执行任何操作（当我在没有上述代码的情况下调用函数本身时，它工作正常）。所以我认为我使用了多重处理错误（尽管我从网站上获取了示例），有什么建议吗？

更新：我注意到当它用 control-c 冻结时我什至无法摆脱它......它总是可以摆脱我的有缺陷的程序。我查看了 python2.5 多处理池 并尝试遵循建议并将导入添加到我的if 语句但没有运气

更新2：对不起，刚刚意识到感谢下面的答案，该命令有效，但它似乎并没有终止程序或让我强制退出。

Answer 1

多处理不是线程。

您可能正在做这样的事情：

data = {}

def new_awesome_function(a, b):
    data[a] = b

运行脚本后，数据没有改变。这是因为多处理使用程序的副本。您的函数正在运行，但它们是在您的程序的副本中运行的，因此对您的原始程序没有影响。

为了利用多处理，您需要显式地从一个进程到另一个进程进行通信。对于线程来说，一切都是共享的，但是对于多处理来说，除非您明确共享，否则什么都不会共享。

最简单的方法是使用返回值：

def new_awesome_function(a, b):
    return a + b

result = pool.apply_async(new_awesome_function, (1, 2))
# later...
value = result.get()

请参阅 python 文档： http://docs.python.org /library/multiprocessing.html，用于其他方法，例如队列、管道和管理器。你不能做的是改变你的程序状态并期望它能够工作。

Answer 2

我不知道您正在使用什么数据库，但很可能您无法像这样在进程之间共享数据库连接。

在Linux上，使用fork()，它在启动子进程时复制内存中的所有内容。然而，除非专门设计，否则诸如套接字、打开文件和数据库连接之类的东西将无法正常工作。

在 Window 上，fork() 不可用，因此它将重新运行您的脚本。就你而言，这将非常糟糕，因为它会再次丢弃所有内容。您可以通过放入 if __name__ == '__main__': 位来防止这种情况发生。

您应该能够在 my_awesome_function 中重新打开数据库连接，从而能够成功地与数据库交互。

说实话，这样做你不会获得任何速度。事实上，我预计这会慢一些。看到数据库真的很慢。您的进程将花费大部分时间等待数据库。现在你只有多个进程在等待数据库，这确实不会改善情况。

但数据库是用来存储东西的。只要您正在进行处理，就应该在访问数据库之前在代码中进行处理。您基本上使用数据库作为集合，并且使用 python 集合您的代码会更好。如果您确实需要将这些内容放入数据库中，请在程序末尾执行此操作。

Answer 3

你的代码似乎对我有用：

from multiprocessing import Pool
import time

def new_awesome_function(a,b):
    print(a,b, 'start')
    time.sleep(1)
    print(a,b, 'end')

if __name__ == '__main__':
    data = [1,2,3,4,5]
    pool = Pool(processes=4)
    for i, x in enumerate(data): 
        pool.apply_async(new_awesome_function, (i, x))
    pool.close()
    pool.join()

给了我：

0 1 start
1 2 start
2 3 start
3 4 start
1 2 end
0 1 end
4 5 start
2 3 end
3 4 end
4 5 end

是什么让你认为它不起作用？

---

编辑：尝试运行它并查看输出：

from multiprocessing import Pool
import time

def new_awesome_function(a,b):
    print(a,b, 'start')
    time.sleep(1)
    print(a,b, 'end')
    return a + b

if __name__ == '__main__':
    data = [1,2,3,4,5]
    pool = Pool(processes=4)
    results = []
        for i, x in enumerate(data): 
        r = pool.apply_async(new_awesome_function, (i, x))
        results.append((i,r))
    pool.close()
    already = []
    while len(already) < len(data):
        for i,r in results:
            if r.ready() and i not in already:
                already.append(i)
                print(i, 'is ready!')
    pool.join()

我的是：

0 1 start
1 2 start
2 3 start
3 4 start
0 1 end
4 5 start
1 2 end
2 3 end
0 is ready!
3 4 end
1 is ready!
2 is ready!
3 is ready!
4 5 end
4 is ready!