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

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

data 是一个列表（[1,2,3,4,5]），我正在尝试将列表发送到要在多个 cpu 上完成的每个项目，但是当我将我的工作命令包装到一个函数中并发送这段代码它没有做任何事情（当我在没有上面代码的情况下调用函数本身时它工作正常）。所以我认为我使用错误的多处理（虽然我从网站上举了例子），有什么建议吗？

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

Update2：对不起，感谢下面的答案，该命令有效，但它似乎没有终止程序或让我强制退出。

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!