Python 多处理。同时完成所有流程答案

【问题标题】：Python multiprocessing. Finish all processes at the same timePython 多处理。同时完成所有流程
【发布时间】：2014-07-17 09:37:31
【问题描述】：

我的目标是将一些数据上传到数据库。我正在使用 psycopg2，它有一个规则：所有进程都必须有自己的数据库连接。就我而言，这意味着我必须投入工作。问题是我只能提交所有进程完成 sql insert 命令。我需要什么：

from multiprocessing import Process

def worker1(s):    
    conn = psycopg2.connect("dbname=mydb user=postgres")
    cursor = conn.cursor()

    pg_cursor.execute(
            """ insert into "MyTable1"("Column1")
                values(%s)""", [1])

    #wait all processes
    conn.commit()       

def worker2(s):    
    conn = psycopg2.connect("dbname=mydb user=postgres")
    cursor = conn.cursor()

    pg_cursor.execute(
            """ insert into "MyTable2"("Column1")
                values(%s)""", [1])

    #wait all processes
    conn.commit()

if __name__ == '__main__':
    p1 = Process(target=worker1)
    p2 = Process(target=worker2)
    p1.start()
    p2.start()

如何让所有进程等到完成 sql 命令？这样做的正确方法是什么？此 sql 仅插入示例，在实际任务中我需要插入数百万条记录。

【问题讨论】：

在进程上调用.join()。
如果我没记错的话，加入让主进程等到所有子进程完成。但我需要子进程相互等待。我在worker中展示了我需要暂停子进程的地方。
我提供的示例仅供参考。您必须为您的特定案例制定最佳的事件流程。

标签： python locking multiprocessing semaphore psycopg2

【解决方案1】：

您可以使用一对multiprocessing.Event 对象来允许两个工作人员告诉对方他们已经完成，并强制他们等待对方回复信号：

from multiprocessing import Process, Event

def worker1(s, my_e, other_e):    
    conn = psycopg2.connect("dbname=mydb user=postgres")
    cursor = conn.cursor()

    pg_cursor.execute(
            """ insert into "MyTable1"("Column1")
                values(%s)""", [1])

    #wait all processes
    my_e.set()
    other_e.wait()
    conn.commit()       

def worker2(s, my_e, other_e):    
    conn = psycopg2.connect("dbname=mydb user=postgres")
    cursor = conn.cursor()

    pg_cursor.execute(
            """ insert into "MyTable2"("Column1")
                values(%s)""", [1])

    #wait all processes
    my_e.set()
    other_e.wait()
    conn.commit()

if __name__ == '__main__':
    e1 = Event()
    e2 = Event()
    p1 = Process(target=worker1, args=(e1, e2))
    p2 = Process(target=worker2, args=(e2, e1))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

如果其中一名工作人员由于某种原因失败，请小心出现死锁的可能性。您可能希望将Event.wait 的可选timeout 关键字参数设置为某个相当高的值，并在它过期时回滚。或者在每个工作人员中使用try/except，以保证my_e 得到set()，但也可以在两个工作人员之间共享一个变量，该变量可用于告诉另一个工作人员是否发生故障。

【讨论】：